kafka-stream-processing

Kafka 流处理

用于使用 Apache Kafka 构建事件驱动应用的综合技能。掌握生产者、消费者、Kafka Streams、连接器、Schema Registry，以及用于大规模实时数据处理的生产环境部署模式。

何时使用此技能

在以下情况下使用此技能：

• 构建事件驱动的微服务架构
• 处理实时数据流和事件日志
• 实现发布订阅消息系统
• 为分析和 ETL 创建数据管道
• 使用有状态处理构建流式数据应用
• 使用 Kafka Connect 集成异构系统
• 实现变更数据捕获（CDC）模式
• 构建实时仪表板和监控系统
• 大规模处理 IoT 传感器数据
• 实现事件溯源和 CQRS 模式
• 构建需要保证消息送达的分布式系统
• 创建实时推荐引擎
• 使用精确一次语义处理金融交易
• 构建日志聚合和监控管道

核心概念

Apache Kafka 架构

Kafka 是一个分布式流处理平台，专为高吞吐量、容错的实时数据处理而设计。

关键组件：

1. 主题：用于组织消息的命名类别
2. 分区：主题中有序且不可变的记录序列
3. 代理：存储和提供数据的 Kafka 服务器
4. 生产者：向主题发布记录的应用
5. 消费者：从主题读取记录的应用
6. 消费者组：共同协调并分担工作负载的消费者
7. ZooKeeper/KRaft：集群协调机制（ZooKeeper 为旧方案，KRaft 为现代方案）

设计原则：

Kafka Design Philosophy:
  - High Throughput: Millions of messages per second
  - Low Latency: Single-digit millisecond latency
  - Durability: Replicated, persistent storage
  - Scalability: Horizontal scaling via partitions
  - Fault Tolerance: Automatic failover and recovery
  - Message Delivery Semantics: At-least-once, exactly-once support

主题和分区

主题是数据流的逻辑通道。每个主题都分为多个分区，以实现并行处理和可扩展性。

# Create a topic with 20 partitions and replication factor 3
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my_topic_name \
--partitions 20 --replication-factor 3 --config x=y

分区特性：

• 有序：分区内的消息严格有序
• 不可变：记录写入后无法修改
• 仅追加：新记录追加到分区末尾
• 保留：可按时间或大小配置保留策略
• 复制：每个分区都会复制到多个代理上

添加分区：

# Increase partition count (cannot decrease)
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic my_topic_name \
--partitions 40

注意：添加分区不会重新分配现有数据，并且可能会影响使用自定义分区逻辑的消费者。

流分区和任务

<h3>Stream Partitions and Tasks</h3>
<p> The messaging layer of Kafka partitions data for storing and transporting it. Kafka Streams partitions data for processing it. In both cases, this partitioning is what enables data locality, elasticity, scalability, high performance, and fault tolerance. Kafka Streams uses the concepts of <b>partitions</b> and <b>tasks</b> as logical units of its parallelism model based on Kafka topic partitions. There are close links between Kafka Streams and Kafka in the context of parallelism: </p>
<ul>
<li>Each <b>stream partition</b> is a totally ordered sequence of data records and maps to a Kafka <b>topic partition</b>.</li>
<li>A <b>data record</b> in the stream maps to a Kafka <b>message</b> from that topic.</li>
<li>The <b>keys</b> of data records determine the partitioning of data in both Kafka and Kafka Streams, i.e., how data is routed to specific partitions within topics.</li>
</ul>
<p> An application's processor topology is scaled by breaking it into multiple tasks. More specifically, Kafka Streams creates a fixed number of tasks based on the input stream partitions for the application, with each task assigned a list of partitions from the input streams (i.e., Kafka topics). The assignment of partitions to tasks never changes so that each task is a fixed unit of parallelism of the application. Tasks can then instantiate their own processor topology based on the assigned partitions; they also maintain a buffer for each of its assigned partitions and process messages one-at-a-time from these record buffers. As a result stream tasks can be processed independently and in parallel without manual intervention. </p>
<p> Slightly simplified, the maximum parallelism at which your application may run is bounded by the maximum number of stream tasks, which itself is determined by maximum number of partitions of the input topic(s) the application is reading from. For example, if your input topic has 5 partitions, then you can run up to 5 applications instances. These instances will collaboratively process the topic's data. If you run a larger number of app instances than partitions of the input topic, the "excess" app instances will launch but remain idle; however, if one of the busy instances goes down, one of the idle instances will resume the former's work. </p>
<p> It is important to understand that Kafka Streams is not a resource manager, but a library that "runs" anywhere its stream processing application runs. Multiple instances of the application are executed either on the same machine, or spread across multiple machines and tasks can be distributed automatically by the library to those running application instances. The assignment of partitions to tasks never changes; if an application instance fails, all its assigned tasks will be automatically restarted on other instances and continue to consume from the same stream partitions. </p>

消息送达语义

Design:
  - The Producer: Design considerations.
  - The Consumer: Design considerations.
  - Message Delivery Semantics: At-least-once, at-most-once, exactly-once.
  - Using Transactions for atomic operations.

至少一次送达：

• 生产者持续重试，直到收到确认
• 消费者在处理后提交偏移量
• 风险：故障时会重复处理
• 适用场景：允许重复处理，或采用幂等处理时

至多一次送达：

• 消费者在处理前提交偏移量
• 生产者不重试
• 风险：故障时会丢失消息
• 适用场景：允许数据丢失时（例如，指标数据）

精确一次语义（EOS）：

• 使用幂等生产者进行事务性写入
• 消费者仅读取已提交的消息
• 适用场景：金融交易、关键数据处理

生产者负载均衡

ProducerClient:
  publish(topic: str, message: bytes, partition_key: Optional[str] = None)
    topic: The topic to publish the message to.
    message: The message payload to send.
    partition_key: Optional key to determine the partition. If None, random partitioning is used.

  get_metadata(topic: str) -> dict
    topic: The topic to get metadata for.
    Returns: A dictionary containing broker information and partition leader details.

生产者无需经过路由层，即可将数据发送到分区领导者代理。Kafka 节点会向生产者提供元数据，以便生产者将请求发送到正确的分区领导者。生产者可以实现自定义分区逻辑，也可以采用随机分配。

生产者

Kafka 生产者会根据可配置的可靠性和性能特性向主题发布记录。

生产者 API

Producer API:
  - send(record): Sends a record to a Kafka topic.
    - Parameters:
      - record: The record to send, including topic, key, and value.
    - Returns: A Future representing the result of the send operation.
  - flush(): Forces any buffered records to be sent.
  - close(): Closes the producer, releasing any resources.
  - metrics(): Returns metrics about the producer.

  Configuration:
  - bootstrap.servers: A list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
  - key.serializer: The serializer class for key that implements the org.apache.kafka.common.serialization.Serializer interface.
  - value.serializer: The serializer class for value that implements the org.apache.kafka.common.serialization.Serializer interface.
  - acks: The number of acknowledgments the producer requires the leader to have received before considering a request complete.
  - linger.ms: The producer groups together any records that arrive in between request transmissions into a single batched request.
  - batch.size: The producer will attempt to batch records together into fewer requests whenever multiple records are being sent to the same partition.

生产者配置

基本设置：

1. bootstrap.servers：Kafka 集群连接字符串

   • 格式：host1:9092,host2:9092,host3:9092
   • 使用多个 Broker 提高容错能力

2. key.serializer / value.serializer：数据序列化

   • org.apache.kafka.common.serialization.StringSerializer
   • org.apache.kafka.common.serialization.ByteArraySerializer
   • 为复杂类型使用自定义序列化器

3. acks：确认级别

   • 0：不需要确认（发送后不管）
   • 1：仅 Leader 确认
   • all / -1：所有同步副本都确认（最高持久性）

4. retries：发送失败时的重试次数

   • 默认值：2147483647 (Integer.MAX_VALUE)
   • 设置为 0 可禁用重试

5. enable.idempotence：精确一次语义

   • true：启用幂等生产者（防止重复）
   • false：默认行为

性能调优：

1. linger.ms：批处理延迟

   • 默认值：0（立即发送）
   • 较高的值（5-100ms）可通过批处理提高吞吐量
   • 权衡：延迟与吞吐量

2. batch.size：批次大小（字节）

   • 默认值：16384 (16KB)
   • 较大的批次可提高吞吐量
   • 每个分区的单个批次最大大小

3. compression.type：消息压缩

   • 选项：none、gzip、snappy、lz4、zstd
   • 减少网络带宽和存储占用
   • 压缩和解压缩会带来 CPU 开销

4. buffer.memory：生产者缓冲区总内存

   • 默认值：33554432 (32MB)
   • 用于缓冲尚未发送的记录
   • 缓冲区满时生产者会阻塞

生产者示例（Java）

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class SimpleProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            ProducerRecord<String, String> record =
                new ProducerRecord<>("my-topic", "key1", "Hello Kafka!");

            // Async send with callback
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("Error producing: " + exception);
                } else {
                    System.out.printf("Sent to partition %d, offset %d%n",
                        metadata.partition(), metadata.offset());
                }
            });
        }
    }
}

生产者最佳实践

1. 使用批处理：配置 linger.ms 和 batch.size 以提高吞吐量
2. 启用幂等性：设置 enable.idempotence=true 以提高可靠性
3. 处理错误：实现适当的回调错误处理
4. 使用压缩：为大型消息启用压缩
5. 分区键：使用有意义的键进行有序处理
6. 关闭生产者：始终在 finally 块中关闭生产者
7. 监控指标：跟踪生产者指标（record-send-rate、compression-rate）
8. 资源池：尽可能重复使用生产者实例

消费者

Kafka 消费者从主题读取记录，支持单独消费和基于消费者组的消费。

消费者组

消费者组支持并行处理，并可自动进行负载均衡和容错。

关键概念：

• Group ID：消费者组的唯一标识符
• 分区分配：组中的每个分区由一个消费者消费
• 再均衡：消费者加入或离开时自动重新分配
• 偏移量管理：消费者组跟踪每个分区已提交的偏移量

消费者组监控：

# List consumer groups
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

# Describe consumer group members with partition assignments
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --members --verbose
CONSUMER-ID HOST CLIENT-ID #PARTITIONS ASSIGNMENT
consumer1-3fc8d6f1-581a-4472-bdf3-3515b4aee8c1 /127.0.0.1 consumer1 2 topic1(0), topic2(0)
consumer4-117fe4d3-c6c1-4178-8ee9-eb4a3954bee0 /127.0.0.1 consumer4 1 topic3(2)
consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1 consumer2 3 topic2(1), topic3(0,1)
consumer3-ecea43e4-1f01-479f-8349-f9130b75d8ee /127.0.0.1 consumer3 0 -

消费者配置

基本设置：

1. bootstrap.servers：Kafka 集群连接
2. group.id：消费者组标识符
3. key.deserializer / value.deserializer：数据反序列化
4. enable.auto.commit：自动提交偏移量
   • true：定期自动提交偏移量
   • false：手动管理偏移量
5. auto.offset.reset：未找到偏移量时的行为
   • earliest：从头开始
   • latest：从末尾开始
   • none：抛出异常

消费者专用的 Kafka Streams 默认值：

Parameter Name: max.poll.records
Corresponding Client: Consumer
Streams Default: 100

Parameter Name: client.id
Corresponding Client: -
Streams Default: <application.id>-<random-UUID>

Parameter Name: enable.auto.commit
Description: Controls whether the consumer automatically commits offsets. When true, the consumer will automatically commit offsets periodically based on the poll interval.
Default Value: true

消费者示例（Java）

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;

public class SimpleConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("my-topic"));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Partition: %d, Offset: %d, Key: %s, Value: %s%n",
                        record.partition(), record.offset(), record.key(), record.value());

                    // Process record
                    processRecord(record);
                }

                // Manual commit after processing batch
                consumer.commitSync();
            }
        }
    }

    private static void processRecord(ConsumerRecord<String, String> record) {
        // Business logic here
    }
}

消费者偏移量管理

偏移量提交策略：

1. 自动提交（默认）：

   • 简单，但对于至少一次投递来说有风险
   • 可能在处理完成前提交

2. 手动同步提交：

   • 阻塞，直到提交成功
   • 确保继续处理前偏移量已提交
   • 吞吐量较低

3. 手动异步提交：

   • 非阻塞提交
   • 吞吐量较高
   • 在回调中处理失败

4. 混合方式：

   • 处理期间进行异步提交
   • 在再均衡或关闭前进行同步提交

// Async commit with callback
consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        System.err.println("Commit failed: " + exception);
    }
});

// Sync commit for reliability
try {
    consumer.commitSync();
} catch (CommitFailedException e) {
    System.err.println("Commit failed: " + e);
}

消费者最佳实践

1. 选择正确的自动提交策略：需要至少一次语义时禁用自动提交
2. 处理重新均衡：实现 ConsumerRebalanceListener
3. 高效处理：尽量缩短 poll() 调用的耗时
4. 平稳关闭：正确关闭消费者
5. 监控滞后：跟踪消费者滞后指标
6. 分区分配：了解分配策略
7. 线程安全：Kafka 消费者不是线程安全的
8. 错误处理：针对瞬时故障实现重试逻辑

Kafka Streams

Kafka Streams 是一个用于构建具备有状态处理能力的实时流式应用的客户端库。

Kafka Streams 架构

处理器拓扑：

There are two special processors in the topology:
<ul>
<li><b>Source Processor</b>: A special type of stream processor that does not have any upstream processors. It produces an input stream to its topology from one or multiple Kafka topics by consuming records from these topics and forwarding them to its down-stream processors.</li>
<li><b>Sink Processor</b>: A special type of stream processor that does not have down-stream processors. It sends any received records from its up-stream processors to a specified Kafka topic.</li>
</ul>
Note that in normal processor nodes other remote systems can also be accessed while processing the current record. Therefore the processed results can either be streamed back into Kafka or written to an external system.

子拓扑：

应用会分解为由重新分区主题连接的子拓扑。每个子拓扑都可以独立扩展。

KStream、KTable 与 GlobalKTable

KStream：不可变的记录流

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Long> wordCounts = builder.stream(
    "word-counts-input-topic", /* input topic */
    Consumed.with(
        Serdes.String(), /* key serde */
        Serdes.Long() /* value serde */
    )
);

KTable：变更日志流（每个键对应最新值）

import org.apache.kafka.streams.StreamsBuilder;

StreamsBuilder builder = new StreamsBuilder();
builder.table("input-topic");

GlobalKTable：可供所有实例使用的完整副本表

KTable: Each application instance gets data from only 1 partition.
GlobalKTable: Each application instance gets data from all partitions.

将流写入 Kafka

KStream<String, Long> stream = ...;
// Write the stream to the output topic, using the configured default key
// and value serdes.
stream.to("my-stream-output-topic");

// Write the stream to the output topic, using explicit key and value serdes,
// (thus overriding the defaults in the config properties).
stream.to("my-stream-output-topic", Produced.with(Serdes.String(), Serdes.Long()));

任何流和表都可以（持续）写回 Kafka 主题。根据具体情况，输出数据可能会重新分区。

重新分区

手动重新分区并指定分区数：

KStream<byte[], String> stream = ... ;
KStream<byte[], String> repartitionedStream = stream.repartition(Repartitioned.numberOfPartitions(10));

Kafka Streams 将生成的主题作为内部主题进行管理，确保清理数据，并支持下游子拓扑扩展。当之前执行了键变更操作且未触发自动重新分区时，这项操作很有用。

联结与协分区

Join Co-partitioning Requirements:

For equi-joins in Kafka Streams, input data must be co-partitioned. This ensures that records with the same key from both sides of the join are delivered to the same stream task.

Requirements for data co-partitioning:
1. Input topics (left and right sides) must have the same number of partitions.
2. All applications writing to the input topics must use the same partitioning strategy to ensure records with the same key are delivered to the same partition number.
   - This applies to producer settings like `partitioner.class` (e.g., `ProducerConfig.PARTITIONER_CLASS_CONFIG`) and Kafka Streams `StreamPartitioner` for operations like `KStream#to()`.
   - Using default partitioner settings across all applications generally satisfies this requirement.

Why co-partitioning is required:
- KStream-KStream, KTable-KTable, and KStream-KTable joins are performed based on record keys (e.g., `leftRecord.key == rightRecord.key`). Co-partitioning by key ensures these records meet.

Exceptions where co-partitioning is NOT required:
1. KStream-GlobalKTable joins:
   - All partitions of the GlobalKTable's underlying changelog stream are available to each KafkaStreams instance.
   - A `KeyValueMapper` allows non-key based joins from KStream to GlobalKTable.
2. KTable-KTable Foreign-Key joins:
   - Kafka Streams internally ensures co-partitioning for these joins.

有状态操作

Kafka Streams 使用状态存储支持聚合、窗口化和联结等有状态操作。

状态存储类型：

• 键值存储：用于聚合和联结
• 窗口存储：用于基于时间的操作
• 会话存储：用于基于会话的聚合

状态存储配置：

Internal Topic Configuration:

- message.timestamp.type: 'CreateTime' for all internal topics.

- Internal Repartition Topics:
  - compaction.policy: 'delete'
  - retention.time: -1 (infinite)

- Internal Changelog Topics for Key-Value Stores:
  - compaction.policy: 'compact'

- Internal Changelog Topics for Windowed Key-Value Stores:
  - compaction.policy: 'delete,compact'
  - retention.time: 24 hours + windowed store setting

- Internal Changelog Topics for Versioned State Stores:
  - cleanup.policy: 'compact'
  - min.compaction.lag.ms: 24 hours + store's historyRetentionMs

应用并行度

The parallelism of a Kafka Streams application is primarily determined by how many partitions the input topics have. For example, if your application reads from a single topic that has ten partitions, then you can run up to ten instances of your applications. You can run further instances, but these will be idle.
The number of topic partitions is the upper limit for the parallelism of your Kafka Streams application and for the number of running instances of your application.
To achieve balanced workload processing across application instances and to prevent processing hotpots, you should distribute data and processing workloads:
Data should be equally distributed across topic partitions. For example, if two topic partitions each have 1 million messages, this is better than a single partition with 2 million messages and none in the other.
Processing workload should be equally distributed across topic partitions. For example, if the time to process messages varies widely, then it is better to spread the processing-intensive messages across partitions rather than storing these messages within the same partition.

Kafka Streams 配置

客户端前缀：

Properties streamsSettings = new Properties();
// same value for consumer, producer, and admin client
streamsSettings.put("PARAMETER_NAME", "value");
// different values for consumer and producer
streamsSettings.put("consumer.PARAMETER_NAME", "consumer-value");
streamsSettings.put("producer.PARAMETER_NAME", "producer-value");
streamsSettings.put("admin.PARAMETER_NAME", "admin-value");
// alternatively, you can use
streamsSettings.put(StreamsConfig.consumerPrefix("PARAMETER_NAME"), "consumer-value");
streamsSettings.put(StreamsConfig.producerPrefix("PARAMETER_NAME"), "producer-value");
streamsSettings.put(StreamsConfig.adminClientPrefix("PARAMETER_NAME"), "admin-value");

特定消费者类型：

Properties streamsSettings = new Properties();
// same config value for all consumer types
streamsSettings.put("consumer.PARAMETER_NAME", "general-consumer-value");
// set a different restore consumer config. This would make restore consumer take restore-consumer-value,
// while main consumer and global consumer stay with general-consumer-value
streamsSettings.put("restore.consumer.PARAMETER_NAME", "restore-consumer-value");
// alternatively, you can use
streamsSettings.put(StreamsConfig.restoreConsumerPrefix("PARAMETER_NAME"), "restore-consumer-value");

主题配置：

Properties streamsSettings = new Properties();
// Override default for both changelog and repartition topics
streamsSettings.put("topic.PARAMETER_NAME", "topic-value");
// alternatively, you can use
streamsSettings.put(StreamsConfig.topicPrefix("PARAMETER_NAME"), "topic-value");

Streams 中的恰好一次语义

Producer Client ID Naming Schema:

  - at-least-once (default):
    `[client.Id]-StreamThread-[sequence-number]`

  - exactly-once (EOS version 1):
    `[client.Id]-StreamThread-[sequence-number]-[taskId]`

  - exactly-once-beta (EOS version 2):
    `[client.Id]-StreamThread-[sequence-number]`

Where `[client.Id]` is either set via Streams configuration parameter `client.id` or defaults to `[application.id]-[processId]` (`[processId]` is a random UUID).

EOS 配置：

Parameter Name: isolation.level
Corresponding Client: Consumer
Streams Default: READ_COMMITTED

Parameter Name: enable.idempotence
Corresponding Client: Producer
Streams Default: true

Parameter Name: transaction.timeout.ms
Corresponding Client: Producer
Streams Default: 10000

Parameter Name: delivery.timeout.ms
Corresponding Client: Producer
Streams Default: Integer.MAX_VALUE

拓扑命名与稳定性

默认拓扑（自动生成名称）：

Topologies: Sub-topology: 0
Source: KSTREAM-SOURCE-0000000000 (topics: [input]) --> KSTREAM-FILTER-0000000001
Processor: KSTREAM-FILTER-0000000001 (stores: []) --> KSTREAM-MAPVALUES-0000000002
<-- KSTREAM-SOURCE-0000000000
Processor: KSTREAM-MAPVALUES-0000000002 (stores: []) --> KSTREAM-SINK-0000000003
<-- KSTREAM-FILTER-0000000001
Sink: KSTREAM-SINK-0000000003 (topic: output)
<-- KSTREAM-MAPVALUES-0000000002

显式命名以确保稳定性：

Kafka Streams Topology Naming:

- Aggregation repartition topics: Grouped
- KStream-KTable Join repartition topic: Joined
- KStream-KStream Join repartition topics: StreamJoined
- KStream-KTable Join state stores: Joined
- KStream-KStream Join state stores: StreamJoined
- State Stores (for aggregations and KTable-KTable joins): Materialized
- Stream/Table non-stateful operations: Named

Operation							Naming Class
------------------------------------------------------------------
Aggregation repartition topics			Grouped
KStream-KStream Join repartition topics		StreamJoined
KStream-KTable Join repartition topic		Joined
KStream-KStream Join state stores		StreamJoined
State Stores (for aggregations and KTable-KTable joins)	Materialized
Stream/Table non-stateful operations		Named

强制使用显式命名：

Properties props = new Properties();
props.put(StreamsConfig.ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG, true);

这可以防止应用使用自动生成的名称启动，从而确保拓扑更新前后保持稳定。

拓扑优化

"topology.optimization":"all"

"topology.optimization":"none"

拓扑优化允许将源主题重用为变更日志主题；从 KStreamBuilder 迁移到 StreamsBuilder 时，这一点至关重要。

使用拓扑的 WordCount 示例

$ mvn clean package
$ mvn exec:java -Dexec.mainClass=myapps.WordCount
Sub-topologies:
Sub-topology: 0
  Source: KSTREAM-SOURCE-0000000000(topics: streams-plaintext-input) --> KSTREAM-FLATMAPVALUES-0000000001
  Processor: KSTREAM-FLATMAPVALUES-0000000001(stores: \[\]) --> KSTREAM-KEY-SELECT-0000000002 <-- KSTREAM-SOURCE-0000000000
  Processor: KSTREAM-KEY-SELECT-0000000002(stores: \[\]) --> KSTREAM-FILTER-0000000005 <-- KSTREAM-FLATMAPVALUES-0000000001
  Processor: KSTREAM-FILTER-0000000005(stores: \[\]) --> KSTREAM-SINK-0000000004 <-- KSTREAM-KEY-SELECT-0000000002
  Sink: KSTREAM-SINK-0000000004(topic: counts-store-repartition) <-- KSTREAM-FILTER-0000000005
Sub-topology: 1
  Source: KSTREAM-SOURCE-0000000006(topics: counts-store-repartition) --> KSTREAM-AGGREGATE-0000000003
  Processor: KSTREAM-AGGREGATE-0000000003(stores: \[counts-store\]) --> KTABLE-TOSTREAM-0000000007 <-- KSTREAM-SOURCE-0000000006
  Processor: KTABLE-TOSTREAM-0000000007(stores: \[\]) --> KSTREAM-SINK-0000000008 <-- KSTREAM-AGGREGATE-0000000003
  Sink: KSTREAM-SINK-0000000008(topic: streams-wordcount-output) <-- KTABLE-TOSTREAM-0000000007
Global Stores: none

此拓扑展示了两个互不相连的子拓扑、它们的源、处理器、接收器，以及用于按聚合键对数据进行混洗的重新分区主题（counts-store-repartition）。

Schema Registry

Schema Registry 在生产者与消费者之间强制执行数据契约，确保数据完整性并防止格式错误的事件。

Data Contracts with Schema Registry:
  - Purpose: Ensure events written to Kafka can be read properly and prevent malformed events.
  - Implementation: Deploy a schema registry alongside the Kafka cluster.
  - Functionality: Manages event schemas and maps them to topics, guiding producers on correct event formats.
  - Note: Kafka does not include a schema registry; third-party implementations are available.

Schema Registry 的优势

1. 架构演进：管理随时间推移发生的架构变更
2. 兼容性检查：强制执行向后／向前兼容性
3. 集中管理：为架构提供唯一可信来源
4. 类型安全：编译时类型检查
5. 文档：自动生成架构文档
6. 版本管理：跟踪每个主题的架构版本

架构格式

支持的格式：

• Avro：具备丰富架构演进能力的紧凑二进制格式
• JSON Schema：易于阅读，并支持架构验证
• Protobuf：Google 的 Protocol Buffers

架构演进兼容模式

1. BACKWARD：新架构可以读取旧数据
2. FORWARD：旧架构可以读取新数据
3. FULL：同时具备向后和向前兼容性
4. NONE：不进行兼容性检查

Avro 生产者示例

import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.*;

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
props.put("schema.registry.url", "http://localhost:8081");

String userSchema = "{"
    + "\"type\":\"record\","
    + "\"name\":\"User\","
    + "\"fields\":["
    + "  {\"name\":\"name\",\"type\":\"string\"},"
    + "  {\"name\":\"age\",\"type\":\"int\"}"
    + "]}";

Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(userSchema);

GenericRecord user = new GenericData.Record(schema);
user.put("name", "John Doe");
user.put("age", 30);

ProducerRecord<String, GenericRecord> record =
    new ProducerRecord<>("users", "user1", user);

producer.send(record);

Kafka Connect

Kafka Connect 是一个用于在 Kafka 和外部系统之间流式传输数据的框架。

Kafka Connect Sink Connector Input Topics

Configuration options for sink connectors to specify input topics using a comma-separated list or a regular expression.

topics
topics.regex

连接器类型

源连接器：将数据导入 Kafka

• 数据库 CDC（Debezium）
• 文件系统
• 消息队列
• 云服务（S3、BigQuery）
• API 和 Webhook

接收器连接器：从 Kafka 导出数据

• 数据库（JDBC、Elasticsearch）
• 数据仓库
• 对象存储
• 搜索引擎
• 分析平台

连接器配置

源连接器示例（JDBC）：

{
  "name": "jdbc-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb",
    "connection.user": "postgres",
    "connection.password": "password",
    "table.whitelist": "users,orders",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "postgres-"
  }
}

接收器连接器示例（Elasticsearch）：

{
  "name": "elasticsearch-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "user-events,order-events",
    "connection.url": "http://localhost:9200",
    "type.name": "_doc",
    "key.ignore": "false"
  }
}

变更数据捕获（CDC）

CDC 捕获数据库变更，并实时将其流式传输到 Kafka。

优势：

• 实时数据同步
• 从现有数据库进行事件溯源
• 微服务数据集成
• 零停机迁移

常用 CDC 连接器：

• Debezium（MySQL、PostgreSQL、MongoDB、SQL Server）
• Oracle GoldenGate
• Maxwell's Daemon

主题管理

Kafka Streams Topic Management:

User Topics:
  - Input Topics: Specified via source processors (e.g., StreamsBuilder#stream(), StreamsBuilder#table(), Topology#addSource()).
  - Output Topics: Specified via sink processors (e.g., KStream#to(), KTable.to(), Topology#addSink()).
  - Management: Must be created and managed manually ahead of time (e.g., via topic tools).
  - Sharing: If shared, users must coordinate topic management.
  - Auto-creation: Discouraged due to potential cluster configuration and default topic settings (e.g., replication factor).

Internal Topics:
  - Purpose: Used internally by the application for state stores (e.g., changelog topics).
  - Creation: Created by the application itself.
  - Usage: Only used by the specific stream application.
  - Permissions: Requires underlying clients to have admin permissions on Kafka brokers if security is enabled.
  - Naming Convention: Typically follows '<application.id>-<operatorName>-<suffix>', but not guaranteed for future releases.

主题操作

DESCRIBE_PRODUCERS:
  - Action: Read
  - Resource: Topic

DESCRIBE_TOPIC_PARTITIONS:
  - Action: Describe
  - Resource: Topic

生成测试消息

$ bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic streams-plaintext-input
>all streams lead to kafka
>hello kafka streams
>join kafka summit

监控与指标

常用指标

Metric Name: outgoing-byte-rate
Description: The average number of outgoing bytes sent per second for a node.
Mbean Name Pattern: kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.w]+),node-id=([0-9]+)

Metric Name: outgoing-byte-total
Description: The total number of outgoing bytes sent for a node.
Mbean Name Pattern: kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.w]+),node-id=([0-9]+)

Metric Name: request-rate
Description: The average number of requests sent per second for a node.
Mbean Name Pattern: kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.w]+),node-id=([0-9]+)

Metric Name: request-total
Description: The total number of requests sent for a node.
Mbean Name Pattern: kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.w]+),node-id=([0-9]+)

Metric Name: request-size-avg
Description: The average size of all requests in the window for a node.
Mbean Name Pattern: kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.w]+),node-id=([0-9]+)

Metric Name: request-size-max
Description: The maximum size of any request sent in the window for a node.
Mbean Name Pattern: kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.w]+),node-id=([0-9]+)

Metric Name: incoming-byte-rate
Description: The average number of incoming bytes received per second for a node.
Mbean Name Pattern: kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.w]+),node-id=([0-9]+)

关键监控领域

1. 生产者指标：

   • record-send-rate
   • record-error-rate
   • compression-rate-avg
   • buffer-available-bytes

2. 消费者指标：

   • records-consumed-rate
   • fetch-latency-avg
   • records-lag-max
   • commit-latency-avg

3. 代理指标：

   • UnderReplicatedPartitions
   • OfflinePartitionsCount
   • ActiveControllerCount
   • RequestHandlerAvgIdlePercent

4. Streams 指标：

   • process-rate
   • process-latency-avg
   • commit-rate
   • poll-rate

生产环境部署

集群架构

多代理设置：

1. 代理：通常至少需要 3 个代理以实现容错
2. 复制：生产环境的复制因子设为 3
3. 分区：分区越多，并行度越高
4. ZooKeeper/KRaft：仲裁需要 3 或 5 个节点

高可用配置

代理配置：

# Broker ID
broker.id=1

# Listeners
listeners=PLAINTEXT://broker1:9092,SSL://broker1:9093

# Log directories (use multiple disks)
log.dirs=/data/kafka-logs-1,/data/kafka-logs-2

# Replication
default.replication.factor=3
min.insync.replicas=2

# Leader election
unclean.leader.election.enable=false
auto.leader.rebalance.enable=true

# Log retention
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

符合条件的 Leader 副本（ELR）

API: DescribeTopicPartitions

Purpose: Fetches detailed information about topic partitions, including Eligible Leader Replicas (ELR).

Usage:
- Via Admin Client: The admin client can fetch ELR info by describing topics.
- Direct API Call: Use the DescribeTopicPartitions API endpoint.

ELR Selection Logic:
- If ELR is not empty, select a replica that is not fenced.
- Select the last known leader if it is unfenced, mimicking pre-4.0 behavior when all replicas are offline.

Dependencies/Side Effects:
- Updating `min.insync.replicas` for a topic will clean the ELR field for that topic.
- Updating the cluster default `min.insync.replicas` will clean ELR fields for all topics.

Return Values:
- ELR status and related replica information for partitions.

安全配置

SSL/TLS 加密：

# SSL configuration
listeners=SSL://broker:9093
security.inter.broker.protocol=SSL
ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
ssl.keystore.password=password
ssl.key.password=password
ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
ssl.truststore.password=password
ssl.client.auth=required

SASL 身份验证：

# SASL/PLAIN configuration
listeners=SASL_SSL://broker:9093
security.inter.broker.protocol=SASL_SSL
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

# JAAS configuration
listener.name.sasl_ssl.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
   username="admin" \
   password="admin-secret" \
   user_admin="admin-secret" \
   user_alice="alice-secret";

性能调优

代理调优：

# Network threads
num.network.threads=8

# I/O threads
num.io.threads=16

# Socket buffer sizes
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600

# Replication
num.replica.fetchers=4
replica.fetch.max.bytes=1048576

# Log flush (rely on OS page cache)
log.flush.interval.messages=9223372036854775807
log.flush.interval.ms=null

吞吐量导向的生产者调优：

acks=1
linger.ms=100
batch.size=65536
compression.type=lz4
buffer.memory=67108864
max.in.flight.requests.per.connection=5

消费者调优：

fetch.min.bytes=1
fetch.max.wait.ms=500
max.partition.fetch.bytes=1048576
max.poll.records=500
session.timeout.ms=30000
heartbeat.interval.ms=3000

最佳实践

生产者最佳实践

1. 启用幂等性：防止消息重复
2. 正确配置 Acks：平衡持久性与吞吐量
3. 使用压缩：降低网络和存储成本
4. 批量发送消息：配置 linger.ms 和 batch.size
5. 处理重试：实现适当的重试逻辑
6. 监控指标：跟踪发送速率和错误率
7. 分区策略：使用有意义的键来确保顺序
8. 正常关闭：使用超时调用 close()

消费者最佳实践

1. 手动管理偏移量：实现至少一次语义
2. 处理再平衡：实现 ConsumerRebalanceListener
3. 缩短轮询时长：高效处理
4. 监控消费者滞后：滞后过高时发出警报
5. 线程安全：每个线程使用一个消费者
6. 正常关闭：正确关闭消费者
7. 错误处理：重试瞬时故障，永久性故障则发送到 DLQ
8. 定位能力：在重放场景中使用 seek()

Kafka Streams 最佳实践

1. 显式命名：使用 Named、Grouped、Materialized 以确保稳定性
2. 状态存储管理：正确配置变更日志主题
3. 错误处理：实现 ProductionExceptionHandler
4. 扩展：使应用实例数与输入分区数相匹配
5. 测试：使用 TopologyTestDriver 进行单元测试
6. 监控：跟踪滞后、处理速率和错误率
7. 恰好一次：为关键应用启用
8. 正常关闭：妥善处理信号

主题设计最佳实践

1. 分区数：根据吞吐量要求确定
2. 副本因子：生产环境主题使用 3
3. 保留策略：根据使用场景（时间或大小）设置
4. 压缩：用于变更日志和查找主题
5. 命名规范：采用一致的命名方案
6. 文档：记录主题用途和架构
7. 访问控制：实施适当的 ACL
8. 监控：跟踪分区指标

运维最佳实践

1. 监控：全面收集指标
2. 告警：针对关键指标设置告警
3. 容量规划：监控磁盘、网络和 CPU
4. 备份：实施灾难恢复策略
5. 升级：经过测试后滚动升级
6. 安全：启用加密和身份验证
7. 文档：维护运行手册
8. 测试：上线前进行负载测试

常见模式

模式 1：事件溯源

将所有状态变更存储为不可变事件：

// Order events
OrderCreated -> OrderPaid -> OrderShipped -> OrderDelivered

// Event store as Kafka topic
Topic: order-events
Compaction: None (keep full history)
Retention: Infinite or very long

模式 2：CQRS（命令查询职责分离）

分离读模型和写模型：

// Write side: Commands produce events
commands -> producers -> events-topic

// Read side: Consumers build projections
events-topic -> streams -> materialized-view (KTable)

模式 3：Saga 模式

协调分布式事务：

// Order saga
order-requested -> payment-requested -> payment-completed ->
inventory-reserved -> order-confirmed

// Compensating transactions on failure
payment-failed -> order-cancelled

模式 4：Outbox 模式

可靠地发布数据库变更：

// Database transaction writes to outbox table
BEGIN TRANSACTION;
  INSERT INTO orders VALUES (...);
  INSERT INTO outbox VALUES (event_data);
COMMIT;

// CDC connector reads outbox and publishes to Kafka
Debezium -> outbox-topic -> downstream consumers

模式 5：扇出模式

将事件广播给多个消费者：

// Single topic, multiple consumer groups
user-events topic
  -> email-service (consumer group: email)
  -> analytics-service (consumer group: analytics)
  -> notification-service (consumer group: notifications)

模式 6：死信队列（DLQ）

处理失败的操作：

try {
  processRecord(record);
} catch (RetriableException e) {
  // Retry
  retry(record);
} catch (NonRetriableException e) {
  // Send to DLQ
  sendToDLQ(record, e);
}

模式 7：窗口聚合

基于时间的聚合：

KStream<String, PageView> views = ...;

// Tumbling window: non-overlapping fixed windows
views.groupByKey()
     .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
     .count();

// Hopping window: overlapping windows
views.groupByKey()
     .windowedBy(TimeWindows.of(Duration.ofMinutes(5))
                            .advanceBy(Duration.ofMinutes(1)))
     .count();

// Session window: activity-based windows
views.groupByKey()
     .windowedBy(SessionWindows.with(Duration.ofMinutes(30)))
     .count();

故障排查

常见问题

问题：消费者滞后不断增加

• 检查消费者处理时间
• 扩展消费者组（添加实例）
• 优化处理逻辑
• 适当时增加 max.poll.records

问题：消息未到达

• 检查生产者 send() 错误回调
• 确认主题存在且可访问
• 检查网络连接
• 查看代理日志中的错误

问题：消息重复

• 启用幂等生产者
• 实现幂等的消费者处理逻辑
• 检查偏移量提交策略
• 确认恰好一次配置

问题：重新平衡耗时过长

• 减少 max.poll.interval.ms
• 增加 session.timeout.ms
• 优化 poll() 处理时间
• 检查消费者健康状况

问题：分区领导者不可用

• 检查代理健康状况和日志
• 确认复制状态
• 检查代理之间的网络
• 查看 ISR（同步副本）

问题：内存不足错误

• 减少 batch.size 和 buffer.memory
• 调整 JVM 堆设置
• 监控内存使用情况
• 检查处理过程中是否存在内存泄漏

迁移和升级策略

Kafka Streams 迁移

从 KStreamBuilder 迁移到 StreamsBuilder：

kstream.repartition(...);
// or for user-managed topics:
kstream.to("user-topic");
streamsBuilder.stream("user-topic");

用于替代 KStream.through() 管理主题重新分区。

主题前缀配置：

Properties props = new Properties();
props.put(StreamsConfig.topicPrefix("my-prefix.") + "replication.factor", "3");
KafkaStreams streams = new KafkaStreams(topology, props);

滚动升级

1. 准备：在预发布环境测试新版本
2. 升级代理：每次升级一个代理
3. 验证：每升级一个代理后检查集群健康状况
4. 升级客户端：生产者、消费者、流处理应用
5. 监控：在整个过程中持续关注指标

资源和参考资料

• Apache Kafka 文档：https://kafka.apache.org/documentation/
• Confluent Platform：https://docs.confluent.io/
• Kafka Streams 文档：https://kafka.apache.org/documentation/streams/
• Schema Registry：https://docs.confluent.io/platform/current/schema-registry/
• Kafka Connect：https://kafka.apache.org/documentation/#connect

---

技能版本：1.0.0 最后更新：2025 年 10 月 技能类别：流处理、事件驱动架构、实时数据 兼容版本：Apache Kafka 2.x、3.x、Confluent Platform