kafka-stream-processing

Kafka ストリーム処理

Apache Kafka を使用してイベント駆動型アプリケーションを構築するための包括的なスキルです。プロデューサー、コンシューマー、Kafka Streams、コネクター、スキーマレジストリ、本番環境へのデプロイパターンを習得し、大規模なリアルタイムデータ処理を実現します。

このスキルを使用するタイミング

次のような場合に、このスキルを使用します。

• イベント駆動型のマイクロサービスアーキテクチャを構築する
• リアルタイムのデータストリームやイベントログを処理する
• パブリッシュ／サブスクライブ型のメッセージングシステムを実装する
• 分析や ETL 用のデータパイプラインを作成する
• ステートフル処理を備えたストリーミングデータアプリケーションを構築する
• Kafka Connect を使用して異種システムを統合する
• 変更データキャプチャ（CDC）パターンを実装する
• リアルタイムのダッシュボードや監視システムを構築する
• IoT センサーデータを大規模に処理する
• イベントソーシングおよび CQRS パターンを実装する
• メッセージの確実な配信が必要な分散システムを構築する
• リアルタイムのレコメンデーションエンジンを作成する
• 厳密に 1 回のセマンティクスで金融取引を処理する
• ログ集約および監視パイプラインを構築する

コアコンセプト

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.

少なくとも 1 回の配信：

• 確認応答を受信するまでプロデューサーが再試行します
• 処理後にコンシューマーがオフセットをコミットします
• リスク：障害発生時に処理が重複します
• 使用例：重複が許容される場合、または冪等な処理を行う場合

最大 1 回の配信：

• 処理前にコンシューマーがオフセットをコミットします
• プロデューサーは再試行しません
• リスク：障害発生時にメッセージが失われます
• 使用例：データ損失が許容される場合（例：メトリクス）

厳密に 1 回のセマンティクス（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
   • 耐障害性を高めるため、複数のブローカーを使用します

2. key.serializer / value.serializer：データのシリアライズ

   • org.apache.kafka.common.serialization.StringSerializer
   • org.apache.kafka.common.serialization.ByteArraySerializer
   • 複雑な型にはカスタムシリアライザーを使用します

3. acks：確認応答レベル

   • 0：確認応答なし（送信後、応答を待たない）
   • 1：リーダーのみが確認応答
   • all / -1：同期中のすべてのレプリカが確認応答（最も高い耐久性）

4. retries：送信に失敗した場合の再試行回数

   • デフォルト：2147483647 (Integer.MAX_VALUE)
   • 再試行を無効にするには、0 に設定します

5. enable.idempotence：厳密に1回のセマンティクス

   • 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 コンシューマーはトピックからレコードを読み取り、個別での消費とグループベースでの消費の両方に対応します。

コンシューマーグループ

コンシューマーグループを使用すると、自動的な負荷分散と耐障害性を備えた並列処理が可能になります。

主な概念：

• グループ ID：コンシューマーグループの一意な識別子
• パーティションの割り当て：グループ内の各パーティションは、1つのコンシューマーによって消費されます
• リバランス：コンシューマーの参加時または離脱時に、自動的に再割り当てされます
• オフセット管理：グループはパーティションごとにコミット済みオフセットを追跡します

コンシューマーグループの監視：

# 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. 自動コミット（デフォルト）：

   • シンプルですが、少なくとも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. 適切な自動コミットを選択する：少なくとも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におけるExactly-Onceセマンティクス

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

このトポロジーには、接続されていない2つのサブトポロジー、それぞれのソース、プロセッサー、シンク、および集約キーでデータをシャッフルするために使用される再パーティション化トピック（counts-store-repartition）が示されています。

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

適格リーダーレプリカ（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. オフセットを手動管理する：少なくとも 1 回の配信セマンティクスを実現します
2. リバランスを処理する：ConsumerRebalanceListener を実装します
3. ポーリング時間を最小限にする：効率的に処理します
4. コンシューマーラグを監視する：ラグが大きい場合にアラートを設定します
5. スレッドセーフティ：スレッドごとにコンシューマーを 1 つ使用します
6. 適切にシャットダウンする：コンシューマーを正しく閉じます
7. エラー処理：一時的な障害は再試行し、恒久的な障害には DLQ を使用します
8. シーク機能：再生が必要な場合は seek() を使用します

Kafka Streams のベストプラクティス

1. 明示的な命名：安定性のために Named、Grouped、Materialized を使用します
2. ステートストアの管理：変更ログトピックを適切に設定します
3. エラー処理：ProductionExceptionHandler を実装します
4. スケーリング：アプリケーションのインスタンス数を入力パーティション数に合わせます
5. テスト：単体テストには TopologyTestDriver を使用します
6. モニタリング：ラグ、処理レート、エラーレートを追跡します
7. Exactly-Once：重要なアプリケーションでは有効にします
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() エラーコールバックを確認します
• トピックが存在し、アクセス可能であることを確認します
• ネットワーク接続を確認します
• ブローカーログでエラーを確認します

問題：メッセージが重複する

• 冪等性プロデューサーを有効にします
• 冪等性のあるコンシューマー処理を実装します
• オフセットのコミット方式を確認します
• Exactly-once の設定を確認します

問題：リバランスに時間がかかりすぎる

• 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. ブローカーのアップグレード：一度に 1 台ずつアップグレードします
3. 確認：各ブローカーのアップグレード後にクラスターの健全性を確認します
4. クライアントのアップグレード：プロデューサー、コンシューマー、Streams アプリをアップグレードします
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