Confluent Replicatorで実現する商用Kafkaレプリケーション

Confluent Replicatorは、Kafka Connect上で動作する商用のレプリケーションコネクタです。データセンター間DRや段階的移行に強く、安定した運用とサポートを求める現場で使われます。

本稿は、公式ドキュメントの挙動に沿って、設計・設定・運用の実務要点をまとめ、CCAAK（Confluent Certified Administrator for Apache Kafka）で問われやすい論点も押さえます。

Replicatorの概要と主なユースケース

Replicatorは、ソースKafkaクラスターからデスティネーションクラスターへトピックを継続的に複製します。Kafka Connectのソースコネクタとして実装されており、スケールアウト、フォールトトレランス、再平衡といったConnectの枠組みをそのまま活用できます。

代表的な用途は、DRサイトへの一方向レプリケーション、データセンター統合時の段階的移行、分析系とプロダクション系の分離（読み取り負荷の隔離）です。スキーマやACLなどの周辺要素は別途ツールの役割分担があるため、データ本体の安定レプリケーションに集中させる設計が無難です。

• 商用サポートとConnectベースの運用基盤を両立
• トピック選択は正規表現やホワイトリストで柔軟に指定可能
• トピック名リネームで段階的カットオーバーを安全に実施
• 重複最小化にはKafkaのidempotent producerとリトライ制御を併用

Connector作成時のREST呼び出し（雰囲気）

POST /connectors
Content-Type: application/json

{
  "name": "replicator-orders",
  "config": {
    "connector.class": "io.confluent.connect.replicator.ReplicatorSourceConnector",
    "tasks.max": "4",
    "src.kafka.bootstrap.servers": "src1:9092,src2:9092",
    "dest.kafka.bootstrap.servers": "dst1:9092,dst2:9092",
    "topic.regex": "^(orders|payments).*",
    "topic.rename.format": "${topic}.dr",
    "key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "provenance.header.enable": "true"
  }
}

アーキテクチャと稼働パターン

Replicatorは、Kafka Connectワーカー上で動作し、ソースに対してはコンシューマ、デスティネーションに対してはプロデューサとして振る舞います。ネットワークは通常、Replicatorノードから両クラスターへ到達可能である必要があります。

運用上は、デスティネーションクラスター側にConnectクラスタを置く構成が一般的です。ソースからの送出規制が厳しい環境でも、Replicatorがソースに読み取り接続できれば一方向コピーを実現できます。

• Connectクラスタは内部トピック（configs/offsets/status）を冗長化
• タスク並列度はパーティション数とネットワーク帯域で調整
• Rebalance時の一時的な遅延スパイクを前提設計に入れる

Replicatorの論理構成

Connectワーカーの基本設定（抜粋）

# worker.properties（抜粋）
bootstrap.servers=dst1:9092,dst2:9092
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
config.storage.replication.factor=3
offset.storage.replication.factor=3
status.storage.replication.factor=3
# 冪等プロデューサで重複を最小化（ソース→デスティ間の書き込み側）
producer.enable.idempotence=true
# セキュリティ例（ワーカー→デスティ）
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="connect" password="secret";

設定の要点とベストプラクティス

トピック選択は正規表現を使い、意図しない新規トピックの流入を防ぐために、命名規約を前提にしたパターンを明示します。段階的移行ではtopic.rename.formatで複製先トピック名を変え、コンシューマの切替を計画的に行います。

コンバータはByteArrayConverterを推奨します。ソースとデスティネーションの間でスキーマ変換を行わないことで、データそのものの忠実なコピーを担保できます。

• topic.regexで範囲を管理、否定先読み等は避けて可読性重視
• key/valueはByteArrayConverterで透過コピー
• tasks.maxはパーティション総数と同等か少し小さめから開始
• エラートピックとDLQの設計（Connectレベルのerror.tolerance等）

Replicatorコネクタ設定（セキュリティ込みの例）

{
  "name": "replicator-secure",
  "config": {
    "connector.class": "io.confluent.connect.replicator.ReplicatorSourceConnector",
    "tasks.max": "6",
    "src.kafka.bootstrap.servers": "src-a:9093,src-b:9093",
    "src.kafka.security.protocol": "SASL_SSL",
    "src.kafka.sasl.mechanism": "PLAIN",
    "src.kafka.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username='src-user' password='src-pass';",
    "dest.kafka.bootstrap.servers": "dst-a:9093,dst-b:9093",
    "dest.kafka.security.protocol": "SASL_SSL",
    "dest.kafka.sasl.mechanism": "PLAIN",
    "dest.kafka.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username='dst-user' password='dst-pass';",
    "topic.regex": "^(orders_|inventory_).*

      

        
        NicheeLab を読み込み中…
      
    
  quot;,
    "topic.rename.format": "${topic}.dr",
    "key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "provenance.header.enable": "true"
  }
}

可用性・順序性・スループットの設計ポイント

順序性はKafkaのパーティション内で保証されます。Replicatorも同様に、同一キーを同一パーティションにルーティングするため、ソースと同数パーティションを基本に設計します。

重複は障害時のリトライで発生し得ます。idempotent producerを有効化し、送信側のretriesやdelivery.timeout.msを適切に設定することで、実務上の重複を最小化できます。

• パーティション整合（ソースとデスティのパーティション数/キー分布を揃える）
• スループットはtasks.maxとfetch/produceバッチサイズでチューニング
• ネットワーク帯域と遅延がRPO/RTOに直結するためSLA化

送信側プロデューサ調整（コネクタの追加プロパティ例）

{
  "producer.linger.ms": "20",
  "producer.batch.size": "131072",
  "producer.compression.type": "lz4",
  "producer.enable.idempotence": "true",
  "producer.max.in.flight.requests.per.connection": "5"
}

セキュリティと運用監視

ソース/デスティネーションともにSASL/SSLでの相互認証を基本とし、最小権限のサービスアカウントを用意します。Replicatorはソース側で読み取り、デスティ側で書き込みの権限が必要です。

運用監視は、Connectのタスク状態、ラグ（ソースとデスティの時刻・オフセット差異）、エラーレート、再平衡頻度を押さえます。Confluent Control CenterやJMXメトリクスで可視化できます。

• ACLはソースでREAD、デスティでWRITE/Create、必要に応じてDescribeConfigs
• 証明書と秘密情報はVault等でローテーション設計
• エラー時のDLQや再試行方針をRunbook化

ACL付与例（概念的、実環境に合わせて調整）

# ソース側: 読み取り（コンシューマ）
kafka-acls --authorizer-properties zookeeper.connect=zk-src:2181 \
  --add --allow-principal User:replicator \
  --operation Read --topic 'orders_*' --group replicator-orders

# デスティ側: 書き込みとトピック作成
auth-cli-or-kafka-acls --add --allow-principal User:replicator \
  --operation Write --operation Create --topic 'orders_*.dr'

ReplicatorとMirrorMaker 2/Cluster Linkingの比較と使い分け

Kafkaのレプリケーション手段は複数あります。Apache標準のMirrorMaker 2（MM2）はオープンソースで幅広く使われ、ConfluentのCluster Linkingはブローカー間で直接ミラーを形成する方式です。ReplicatorはConnectベースの枯れた運用と商用サポートを重視するケースに適します。

試験では、要件に応じた選定理由（ネットワーク制約、両端のバージョン/ディストリビューション混在、移行の段階的切替、運用ガバナンス）を言語化できることがポイントになります。

• 一方向・段階移行・運用標準化ならReplicatorが無難
• 低遅延・直接ミラー・最小オーバーヘッドはCluster Linkingが有力
• OSS指向/標準機能優先ならMM2

Replicator停止・切替の最小Runbook（例）

# 1) 対象トピックのレプリケーションラグを監視し0近傍を確認
# 2) コンシューマを一時停止、最終バッチを待機
# 3) Replicatorを一時停止（/connectors/{name}/pause）
# 4) デスティ側の新トピック（例: *.dr）へコンシューマの接続先を切替
# 5) 段階的に旧系を廃止（監視を併走）

問題で確認

よくある質問