---
title: "Kafka Connectに Cloud Data Streaming Clusterをローテーション"
slug: "cdss-example04"
updated: 2026-09-17T09:25:49Z
published: 2026-09-17T09:25:49Z
canonical: "guide.ncloud-docs.com/cdss-example04"
stale: true
---

> ## Documentation Index
> Fetch the complete documentation index at: https://guide.ncloud-docs.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Kafka Connectに Cloud Data Streaming Clusterをローテーション

VPC環境で利用できます。

別途の VPC Serverを作成して Confluent Replicatorを利用し、Kafka Cluster間のデータをリアルタイムでレプリケーションして Kafka Clusterをローテーションする方法について説明します。 このプロセスは Cent Linuxで作成された Clusterを Rocky Linuxが適用された Clusterに変更することをサポートします。

## 事前タスク

このガイドを実行する前に確認する必要があるタスクは、次の通りです。

- 既存の Cloud Data Streaming Service Clusterの BrokerNodeアクセス情報
- 既存と同じ Kafkaバージョンを持つ新規 Cloud Data Streaming Service Clusterの BrokerNodeアクセス情報
- ドキュメントの内容のうち、既存 Clusterは A、新規で作成した Clusterは Bと呼びます。

## Cluster移行手順

新規で作成した Clusterに移行する手順は、次の通りです。

1. 下部のガイドに従って VPC Server作成、Confluent Replicatorインストール/適用、ACG設定を完了します。
2. 新規で作成した Clusterに既存 Clusterの Topicとデータが正常にレプリケーションされるか確認/検証します。
3. Consumerを新規で作成した Clusterに BootstrapUrlを変更します。
4. Producerを新規で作成した Clusterに BootstrapUrlを変更します。
5. 既存 Clusterの Kafka Topicに Queueが完全に処理されているか確認/検証します。
6. Confluent Replicatorがインストールされた VPC Serverと既存クラスタは停止後に返却します。

## VPC Server作成

### STEP 1.サーバ作成

Confluent Replicatorをインストールするサーバを作成します。

1. NAVERクラウドプラットフォームコンソールの VPC環境で、**Menu** > **All Services** > **Compute** > **Server**メニューを順にクリックします。
2. **[サーバ作成]** ボタンをクリックしてサーバを作成します([サーバ作成ガイド](/docs/ja/server-create-vpc))。
3. サーバ作成時に Subnetの種類は Clusterの BrokerNodeと同じか同じタイプ(Private)の Subnetで作成します。
4. 作成されたサーバの SSHでアクセスし、Confluent Replicatorインストールを準備します([サーバアクセスガイド](/docs/ja/server-access-vpc))。
  - private subnetに作成された VPC Serverの場合 NATG/Wが追加必要で、ルーティング設定をする必要があります。

## ネットワーク ACG設定

### STEP 1.ACG設定

新規で作成された VPC Serverは Cloud Data Streaming Service Clusterの BrokerNodeの 9092Portにアクセスできる必要があります。 このために ACGを追加する方法は、次の通りです。

1. NAVERクラウドプラットフォームコンソールの VPC環境で、**Menu** > **All Services** > **Big Data & Analytics** > **Cloud Data Streaming Service**メニューを順にクリックします。
2. Clusterの選択後、ブローカーノード ACGの横にある新しいウィンドウを開くボタンをクリックします。
3. ブローカーノード ACGの名前や ACGIDが同じである ACGを選択した後、上部の **[ACG設定]** ボタンをクリックします。
4. inboudタブで以下の情報を入力した後、 **[追加]** ボタンをクリックします。
  - プロトコル: TCP
  - アクセスソース: VPC Serverの IPアドレスや VPC Serverが属している ACGの名前
  - 許可ポート: 9092
5. 下部の **[適用]** ボタンをクリックします。
6. 同じ方法で現在利用中のクラスタと変更するクラスタのいずれも同じ ACGを登録します。

## Confluent Replicatorのインストール

Serverに Confluent Replicatorをインストールするユースケースを説明します。インストールプロセスで Kafka Broker Nodeに対するアクセス情報を確認する必要があります。Confluent Replicatorは、ライセンスに従って利用可否を確認する必要があり、NAVERクラウドプラットフォームでは利用方法のみをガイドします。インストールプロセスで Kafka Broker Nodeに対するアクセス情報を確認する必要があります。

### STEP 1.Javaのインストール

1. 以下のコマンドを入力して jdkをインストールします  

最小1.8または java11以上をお勧めします。

```
sudo yum install java-devel -y
```

### STEP 2.Confluent Replicatorのインストール

Confluent Replicatorは Confluent Platformの一部として提供します。作成した VPC Serverに Confluent Platformをインストールし、Replicatorコネクタをダウンロードします。

**Confluent Platformのダウンロード** Confluent Platformは Confluentダウンロードページでダウンロードできます。

```
curl -O https://packages.confluent.io/archive/7.5/confluent-7.5.0.tar.gz
```

ダウンロードしたファイルを解凍した後、ディレクトリに移動します。

```
tar -xzf confluent-7.5.0.tar.gz
cd confluent-7.5.0
```

Confluent Platformインストールディレクトリパスを環境変数として設定します(~/.bashrcまたは ~/.bash_profile変更)。

```
export CONFLUENT_HOME=/path/to/confluent-7.5.0
export PATH=$CONFLUENT_HOME/bin:$PATH
```

変更事項を適用します

```
source ~/.bashrc
または
source ~/.bash_profile
```

**Confluent Hubを通じた Replicatorのインストール** Replicatorの最新バージョンコネクタをインストールします。

```
confluent-hub install confluentinc/kafka-connect-replicator:latest
```

### STEP 3.Kafkaクラスタ構成の確認と設定

VPC Serverにインストールされた Replicatorが2つのクラスタのブローカーにアクセスできるか確認します。Replicatorから Kafkaブローカーが実行中のポートにアクセスでき、ACG設定へのアクセスが許可されている必要があります。

- Kafkaトピック確認コマンドの例

```
kafka-topics --bootstrap-server A_CLUSTER_BROKER_1_IP:9092,A_CLUSTER_BROKER_2_IP:9092 --list
kafka-topics --bootstrap-server B_CLUSTER_BROKER_1_IP:9092,B_CLUSTER_BROKER_2_IP:9092 --list
```

Topicを指定して移動させる場合、B Clusterは A Clusterと同じ Topicを持つ必要があります。以下のユースケースは、A Clusterに「example-topic」という名前を持つ Topicがある場合のユースケースです。

- 自動で作成して全体を移動させたい場合、以下の STEP 4.Replicator設定ファイル作成をご参照ください。

```
kafka-topics --bootstrap-server B_CLUSTER_BROKER_1_IP:9092 --create --topic example-topic --partitions 3 --replication-factor 2
```

### STEP 4.Replicator設定ファイル作成

中間サーバで Kafka Replicator設定ファイル(replicator.properties)を作成します。このファイルは2つのクラスタ間のデータのレプリケーションに必要な設定を含みます。/path/to/confluent-7.5.0/etc/kafka-connect-replicator/quickstart-replicator.propertiesファイルを以下の内容に変更します。

```
# コネクタの名前設定
name=replicator-connector

connector.class=io.confluent.connect.replicator.ReplicatorSourceConnector

key.converter=io.confluent.connect.replicator.util.ByteArrayConverter
value.converter=io.confluent.connect.replicator.util.ByteArrayConverter
header.converter=io.confluent.connect.replicator.util.ByteArrayConverter

tasks.max=4

# ソースクラスタの接続情報
src.kafka.bootstrap.servers=A_CLUSTER_BROKER_1_IP:9092,A_CLUSTER_BROKER_2_IP:9092

# 対象クラスタの接続情報
dest.kafka.bootstrap.servers=B_CLUSTER_BROKER_1_IP:9092,B_CLUSTER_BROKER_2_IP:9092

# 対象クラスタにトピックが存在しない場合、自動作成の有無
topic.auto.create=true

# レプリケーションするトピック名(複数の場合 topic1|topic2|topic3可能)
topic.whitelist=example-topic

# レプリケーションせず除外するトピック名
#topic.blacklist=

# レプリケーションされるトピックのトピック名の再定義が必要な場合(既存トピック名.replicaに作成される)
topic.rename.format=${topic}.replica

# トピックのレプリケーションに失敗する場合の再試行間隔設定(10秒)
topic.create.backoff.ms=10000

# オフセット管理の設定
offset.storage.topic=replicator-offsets
config.storage.topic=replicator-configs
status.storage.topic=replicator-status
```

### STEP 5.connectorと Replicator設定ファイル作成

設定ファイルを使用して Kafka Connectを standaloneモードで実行するために connect-standalone.propertiesファイルを変更します。

```
bootstrap.servers=B_CLUSTER_BROKER_1_IP:9092,B_CLUSTER_BROKER_2_IP:9092

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

key.converter.schemas.enable=true
value.converter.schemas.enable=true

offset.storage.file.filename=/tmp/connect.offsets

offset.flush.interval.ms=10000

plugin.path=/usr/share/java,/root/confluent-7.5.0/share/confluent-hub-components
```

### STEP 6.Replicator実行

変更された設定ファイルを適用した kafka Connectを用いて Replicatorを standaloneモードで実行します。 実行すると即時レプリケーションが始まります。

```
connect-standalone $CONFLUENT_HOME/etc/kafka/connect-standalone.properties $CONFLUENT_HOME/etc/kafka-connect-replicator/quickstart-replicator.properties
```

ログファイルは、$CONFLUENT_HOME/logsディレクトリに保存されます。

### STEP 7.トラブルシューティングと追加設定

- アクセスエラー: VPC Serverと Clusterのブローカーノード ACGに 9092 Portが許可されているか確認します。
- データ遅延の問題: レプリケーション速度が遅い場合、consumer.threads値を増やすかブローカーの性能を確認します。
- 複数トピックのレプリケーション: topic.whitelist設定に正規表現式を使用して複数のトピックをレプリケーションしたり、特定のトピックを除外するには topic.blacklistオプションを使用できます

```
topic.whitelist=topic1|topic2|topic3
topic.blacklist=topic4|topic5|topic6
```