Scala 与 Kafka 集成实战：FS2-Kafka、Alpakka 与流式管道

- topic：逻辑主题，可配置 partitions / replication.factor / retention.ms
- partition：有序不可变日志，offset 单调递增，是并行与顺序的最小单元
- offset：消费者位点，__consumer_offsets 内部主题持久化提交结果
- consumer group：组内分区独占，实例数 > 分区数时多出的实例空转
- rebalance：分区再分配；触发条件=成员加入/离开、订阅变化、分区数变化
- 分区键：相同 key 必落同一分区，是「保序」的唯一手段
- ISR：in-sync replicas，acks=all 时需 ISR 全员确认才算写入成功

import org.apache.kafka.clients.admin.{AdminClient, NewTopic}
import org.apache.kafka.common.config.TopicConfig
import java.util.Properties

val props = new Properties()
props.put("bootstrap.servers", "localhost:9092")

val admin = AdminClient.create(props)
val topic = new NewTopic("orders", 12, 3.toShort)
  .configs(Map(
    TopicConfig.RETENTION_MS_CONFIG -> "604800000",
    TopicConfig.CLEANUP_POLICY_CONFIG -> "delete",
    TopicConfig.MIN_INSUFFICIENT_REPLICAS_CONFIG -> "2"
  ).asJava)

admin.createTopics(List(topic).asJava).all().get()

- 依赖：org.typelevel %% fs2-kafka % 3.x（对应 Kafka 3.x 客户端）
- KafkaConsumerSettings：bootstrapServers + groupId + 反序列化器
- ConsumerSettings[F, K, V].withAutoOffsetReset(First/Last) 决定无位点时的起点
- KafkaConsumer.stream(settings).subscribe(topic)：返回 CommittableConsumerRecord 流
- 提交：stream.commitBatchWithin(maxBatch, maxInterval) 或手动 commit
- 生产：KafkaProducer.pipe(producerSettings) 得到 Pipe

import cats.effect.{IO, IOApp}
import fs2.kafka.*
import scala.concurrent.duration.*

object ConsumerApp extends IOApp.Simple {
  val consumerSettings: ConsumerSettings[IO, String, String] =
    ConsumerSettings[IO, String, String]
      .withAutoOffsetReset(AutoOffsetReset.Earliest)
      .withBootstrapServers("localhost:9092")
      .withGroupId("order-processor")
      .withEnableAutoCommit(false)          // 关闭自动提交，手动控制
      .withMaxPollRecords(500)

  val producerSettings: ProducerSettings[IO, String, String] =
    ProducerSettings[IO, String, String]
      .withBootstrapServers("localhost:9092")
      .withEnableIdempotence(true)

  val run: IO[Unit] =
    KafkaConsumer
      .stream(consumerSettings)
      .subscribeTo("orders")
      .records
      .map(_.value.toUpperCase)
      .through(KafkaProducer.pipe(producerSettings))
      .compile
      .drain
}

- at-least-once：处理完成后 commit；崩溃重放 → 下游必须幂等
- at-most-once：commit 后处理；丢数据，仅用于可丢弃的监控类场景
- exactly-once（EOS）：事务性生产者，消费位点写入同一事务
- 幂等生产者：enable.idempotence=true 保证单分区内不重复（重试不产生副本）
- 事务前提：transactional.id 唯一且稳定，max.in.flight <= 5，acks=all
- 跨系统 EOS：Kafka 事务无法覆盖外部 DB → 用 outbox 或幂等键
- 重复来源：rebalance 未提交位点、poll 超时被踢出、处理超 max.poll.interval.ms

import fs2.kafka.*

val transactional: ProducerSettings[IO, String, String] =
  ProducerSettings[IO, String, String]
    .withBootstrapServers("localhost:9092")
    .withTransactionalId("order-tx-1")     // 必须全局唯一且稳定
    .withEnableIdempotence(true)

KafkaProducer
  .transactional(transactional)            // 返回 TransactionalKafkaProducer
  .use { producer =>
    producer.produce(
      ProducerRecords.one(ProducerRecord("orders-out", "k", "v"))
    )
  }

- 串行：evalMap（一次一个 Effect）
- 并行：parEvalMap(n) 保持顺序、parEvalMapUnordered(n) 吞吐优先
- 批量：chunkN(n) / grouped(n) 后 evalMap 一次性写库
- 重试：retry(_.retry) 固定、.delay + retryWithDelay 指数退避
- 超时：.timeout(d) / .timeoutTo(d, fallback)
- 兜底：handleErrorWith 转死信，不要吞异常
- 死信队列：解析/处理失败的消息 produce 到 <topic>.DLQ 并附原始头
- 背压：fs2 天然 pull-based，无需 buffer 调参；buffer(n) 仅平滑抖动

import fs2.{Stream, Pipe}
import scala.concurrent.duration.*

final case class Order(id: String, amount: BigDecimal, raw: String)

def parse(r: CommittableConsumerRecord[IO, String, String]): IO[Either[String, Order]] =
  IO(Order(r.record.key, BigDecimal(r.record.value), r.record.value)).attempt
    .map(_.left.map(_.getMessage))

def pipeline: Pipe[IO, CommittableConsumerRecord[IO, String, String], Unit] =
  _.parEvalMap(8) { rec =>                       // 限并发 8，保序
      parse(rec)
        .timeout(3.seconds)
        .retryWithDelay(3, 500.millis)           // 指数退避重试
        .flatMap {
          case Right(o)  => process(o)
          case Left(err) => toDeadLetter(rec, err)
        }
    }
    .chunkN(100)
    .evalMap(batch => commitBatch(batch))

import akka.actor.ActorSystem
import akka.kafka.scaladsl.{Committer, Consumer, Producer}
import akka.kafka.{ConsumerSettings, ProducerSettings, Subscriptions}
import akka.stream.scaladsl.Source
import org.apache.kafka.clients.consumer.ConsumerConfig

implicit val system: ActorSystem = ActorSystem("kafka")

val consumerSettings = ConsumerSettings(system, new StringDeserializer, new StringDeserializer)
  .withBootstrapServers("localhost:9092")
  .withGroupId("order-processor")
  .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

val producerSettings = ProducerSettings(system, new StringSerializer, new StringSerializer)

val done = Consumer
  .committableSource(consumerSettings, Subscriptions.topics("orders"))
  .mapAsync(8) { msg =>
    process(msg.record.value()).map(_ => msg.committableOffset)
  }
  .groupedWithin(100, 5.seconds)
  .mapAsync(1) { offsets =>
    // 批量提交位点：把 committableOffset 集合交给 Committer
    Source(offsets).runWith(Committer.sink(committerSettings))
  }
  .run()

- JSON + circe：Decoder/Encoder 派生，灵活但无 schema 约束
- Avro：紧凑二进制，Schema Registry 强制兼容性（BACKWARD/FORWARD/FULL）
- Protobuf：强类型、跨语言，适合与 gRPC 共享 .proto
- Schema Registry：subject 命名 <topic>-key / <topic>-value
- 兼容策略：新增字段给默认值=向后兼容；删字段/改类型=破坏性
- 序列化异常处理：DeserializationException 不可重试，直接死信
- 错误处理：KafkaConsumerSettings 可配 withDeserializationExceptionHandler

import io.circe.{Decoder, Encoder}
import io.circe.generic.semiauto.*
import fs2.kafka.*

final case class OrderEvent(id: String, amount: BigDecimal, ts: Long)
object OrderEvent {
  given Encoder[OrderEvent] = deriveEncoder
  given Decoder[OrderEvent] = deriveDecoder
}

val serde: Serde[IO, OrderEvent] =
  Serde.instance(
    serializer   = Serializer.lift[IO, OrderEvent](o => IO.pure(Encoder[OrderEvent](o).noSpaces.getBytes)),
    deserializer = Deserializer.lift[IO, OrderEvent](b =>
      IO.fromEither(io.circe.parser.decode[OrderEvent](new String(b)).left.map(e => new RuntimeException(e.getMessage))))
  )

- 分区键：key = entityId，保证同一实体消息进同一分区
- 累积：mapAccumulate 无界状态，须考虑内存增长
- 窗口：groupWithin(size, time) 做时间+数量双触发
- 状态存储：内存（可重建）或外部（Redis/RocksDB/Postgres）
- 状态容错：Kafka Streams 用 changelog + 本地 RocksDB 恢复
- 幂等：状态更新配合幂等键，避免重放导致重复累加
- 分区再平衡：状态须随分区迁移或可从源重建

import fs2.Stream
import scala.concurrent.duration.*

def windowedCounts[F[_]](
  in: Stream[F, (String, BigDecimal)]
): Stream[F, Map[String, BigDecimal]] =
  in.groupWithin(1000, 5.seconds)          // 1000 条或 5 秒触发
    .map { chunk =>
      chunk.groupBy(_._1).view.mapValues(_.map(_._2).sum).toMap
    }

- consumer lag：kafka_consumergroup_lag（JMX / kafka-exporter / Burrow）
- 分区级 lag：定位是全局慢还是个别分区倾斜
- rebalance 计数：kafka_consumer_rebalance_total 突增说明实例不稳定
- poll/commit：commit 失败率、max.poll.interval 触顶次数
- 处理耗时：p50/p95/p99 直方图，按 topic/partition 维度
- trace 传播：ProducerRecord headers 注入 traceparent，消费侧提取续接
- 告警：lag 持续增长 + rebalance 频繁 = 处理能力不足或处理逻辑变慢

import fs2.kafka.*
import org.apache.kafka.common.header.internals.RecordHeader

val traceId = "0123456789abcdef"
val record = ProducerRecord("orders", "key", "value")
  .withHeaders(Headers(new RecordHeader("traceparent", s"00-$traceId-...-01".getBytes)))

// 消费侧提取
val span = record.headers("traceparent") match {
  case h :: _ => Some(new String(h.value()))
  case Nil    => None
}

- 优雅停机：KillSwitch / 中断信号 → 停止拉取 → 处理在途 → 提交 → 关闭
- RebalanceListener：onPartitionsRevoked 前提交位点（同步提交）
- 消费者并发上限 = 分区数，实例数超过分区数纯属浪费
- 吞吐调优：fetch.min.bytes 增大提升吞吐、max.poll.records 控制单批
- 会话与心跳：session.timeout.ms、heartbeat.interval.ms 需协调
- 测试：Testcontainers KafkaContainer 或 embedded-kafka，覆盖 rebalance 场景
- 版本：客户端版本 >= broker 版本兼容，Kafka 3.x 起 KRaft 免 ZooKeeper

import fs2.Stream
import cats.effect.{IO, Deferred, Resource}

def graceful: Resource[IO, Unit] =
  for {
    stop <- Resource.eval(Deferred[IO, Unit])
    _    <- Resource.make(
              Stream.eval(stop.get).compile.drain.background
            )(_ => stop.complete(()).void)
  } yield ()

// 停机信号到达时：停止拉取 → 排空在途 → commit → close
// Kubernetes: terminationGracePeriodSeconds 必须大于最长批次处理时间