Миграция данных между кластерами Managed Kafka

Вы можете перенести данные из исходного кластера Managed Kafka в целевой с помощью инструмента Apache Kafka MirrorMaker 2 (MM2). Исходный кластер остается доступен для записи во время миграции.

Чтобы мигрировать данные:

Подготовьте виртуальную машину.
Настройте подключение к кластерам.
Проверьте доступ к API Managed Kafka.
Создайте топики в целевом кластере.
Создайте конфигурацию MirrorMaker 2.
Проверьте направление репликации.
Запустите MirrorMaker 2.
Проверьте репликацию.

Если вам больше не нужны созданные ресурсы, удалите их.

Перед началом работы

Создайте исходный и целевой кластеры Managed Kafka.
Создайте пользователя в исходном кластере для чтения данных, а в целевом — для записи.
Создайте сервисный аккаунт sa-kafka-mm2 с ролью mkafka.editor. Аккаунт понадобится для создания топиков в целевом кластере.

Создайте виртуальную машину в той же сети, где расположены оба кластера. При создании ВМ привяжите к ней сервисный аккаунт sa-kafka-mm2. В зависимости от количества переносимых разделов и интенсивности потока данных выберите конфигурацию ВМ:

Параметр	Минимальная конфигурация	Рекомендуемая конфигурация
vCPU	2	от 4
RAM	4 ГБ	от 8 ГБ
Диск	10 ГБ	от 20 ГБ, с запасом под логи

Подготовьте данные для подключений:
- адреса брокеров исходного и целевого кластеров;
- имена и пароли пользователей исходного и целевого кластеров;
- имя проекта;
- IP-адрес ВМ.

1. Подготовьте ВМ

Подключитесь к ВМ по SSH.

Установите необходимые пакеты:

1
sudo apt update && \
2
sudo apt install -y \
3
  curl \
4
  wget \
5
  tar \
6
  gzip \
7
  ca-certificates \
8
  netcat-openbsd \
9
  dnsutils \
10
  jq

Установите Java 17:
bash
```
1
sudo apt install -y openjdk-17-jre-headless
```
Проверьте успешность установки:
bash
```
1
java -version
```

Создайте системного пользователя kafka-mm2 и каталоги для него:

1
sudo useradd \
2
  --system \
3
  --home /opt/kafka \
4
  --shell /usr/sbin/nologin \
5
  kafka-mm2 && \
6
sudo mkdir -p /opt/kafka && \
7
sudo mkdir -p /etc/kafka-mm2 && \
8
sudo mkdir -p /opt/kafka-mm2/scripts

Установите Apache Kafka версии не ниже, чем в кластере:

1
cd /tmp && \
2
wget https://downloads.apache.org/kafka/4.1.2/kafka_2.13-4.1.2.tgz && \
3
tar -xzf kafka_2.13-4.1.2.tgz && \
4
sudo mv kafka_2.13-4.1.2 /opt/kafka/ && \
5
sudo ln -s /opt/kafka/kafka_2.13-4.1.2 /opt/kafka/current && \
6
sudo chown -R kafka-mm2:kafka-mm2 /opt/kafka

Проверьте успешность установки:

1
ls -l /opt/kafka/current/bin/connect-mirror-maker.sh && \
2
ls -l /opt/kafka/current/bin/kafka-topics.sh

Установите корневой сертификат MWS — он понадобится для TLS-подключения к кластерам:

1
cd /tmp && \
2
wget https://pki.mws.ru/certs/mws-root-ca.crt && \
3
sudo mv /tmp/mws-root-ca.crt /etc/kafka-mm2/mws-root-ca.crt && \
4
sudo chown kafka-mm2:kafka-mm2 /etc/kafka-mm2/mws-root-ca.crt && \
5
sudo chmod 644 /etc/kafka-mm2/mws-root-ca.crt

Проверьте доступность кластеров с ВМ. Порт подключения из внутренней сети — 9095:
bash
```
1
nc -vz <адрес брокера исходного кластера> 9095 && \
2
nc -vz <адрес брокера целевого кластера> 9095
```
Ожидаемый результат: обе команды завершаются сообщением succeeded.

2. Настройте подключение к кластерам

Файлы конфигурации содержат параметры подключения к кластеру: аутентификацию SASL_SSL с механизмом SCRAM-SHA-256 и корневой сертификат MWS.

Создайте файл /etc/kafka-mm2/source-client.properties со следующим содержимым:

1
security.protocol=SASL_SSL
2
sasl.mechanism=SCRAM-SHA-256
3

4
ssl.truststore.type=PEM
5
ssl.truststore.location=/etc/kafka-mm2/mws-root-ca.crt
6

7
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="<имя пользователя исходного кластера>" password="<пароль>";
8

9
ssl.endpoint.identification.algorithm=

Создайте файл /etc/kafka-mm2/target-client.properties со следующим содержимым:

1
security.protocol=SASL_SSL
2
sasl.mechanism=SCRAM-SHA-256
3

4
ssl.truststore.type=PEM
5
ssl.truststore.location=/etc/kafka-mm2/mws-root-ca.crt
6

7
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="<имя пользователя целевого кластера>" password="<пароль>";
8

9
ssl.endpoint.identification.algorithm=

Задайте права на файлы конфигурации:

1
sudo chown kafka-mm2:kafka-mm2 \
2
  /etc/kafka-mm2/source-client.properties \
3
  /etc/kafka-mm2/target-client.properties && \
4
sudo chmod 600 \
5
  /etc/kafka-mm2/source-client.properties \
6
  /etc/kafka-mm2/target-client.properties

Проверьте подключение к исходному кластеру:

1
sudo -u kafka-mm2 \
2
/opt/kafka/current/bin/kafka-topics.sh \
3
  --bootstrap-server <адрес брокера исходного кластера> \
4
  --command-config /etc/kafka-mm2/source-client.properties \
5
  --list

Ожидаемый результат: список топиков исходного кластера.

Проверьте подключение к целевому кластеру:
bash
```
1
sudo -u kafka-mm2 \
2
/opt/kafka/current/bin/kafka-topics.sh \
3
  --bootstrap-server <адрес брокера целевого кластера> \
4
  --command-config /etc/kafka-mm2/target-client.properties \
5
  --list
```
Ожидаемый результат: список топиков целевого кластера.

Примечание
Если подключение завершается ошибкой PKIX path building failed, проверьте, что существует файл сертификата MWS /etc/kafka-mm2/mws-root-ca.crt, а в файлах конфигурации указаны параметры ssl.truststore.type=PEM и ssl.truststore.location=/etc/kafka-mm2/mws-root-ca.crt.

3. Проверьте доступ до Managed Kafka API

Топики целевого кластера создаются через API Managed Kafka. Чтобы выполнить запросы, получите IAM-токен сервисного аккаунта sa-kafka-mm2 и проверьте доступ к API с помощью этого токена:

Убедитесь, что привязка сервисного аккаунта к ВМ завершена. Привязка может занять несколько минут:

1
curl -s \
2
  -H "Metadata-Flavor: Google" \
3
  http://169.254.169.254/computeMetadata/v1/instance/service-accounts/?recursive=true \
4
| jq

Ожидаемый вывод:

1
{
2
  "<имя проекта>/serviceAccounts/sa-kafka-mm2": {
3
    "aliases": [
4
      "default"
5
    ]
6
  }
7
}

Получите IAM-токен сервисного аккаунта:

1
TOKEN="$(
2
  curl -s \
3
    -H "Metadata-Flavor: Google" \
4
    http://169.254.169.254/computeMetadata/v1/instance/service-accounts/default/token \
5
  | jq -r '.access_token'
6
)"

MWS автоматически обновляет токен в метаданных ВМ.

Задайте имя проекта и получите список кластеров:

1
export MWS_PROJECT='<имя проекта>' && \
2
curl -sS \
3
  -H "Authorization: Bearer ${TOKEN}" \
4
  "https://mkafka.mwsapis.ru/mkafka/v1/projects/${MWS_PROJECT}/clusters" \
5
| jq -r '
6
  .items[] |
7
  [
8
    (.metadata.id | split("/") | last),
9
    .metadata.displayName
10
  ] | @tsv
11
'

Пример вывода:

1
cluster-source-b    Source Kafka
2
cluster-target-b    Target Kafka

Задайте переменную с техническим именем целевого кластера:
bash
```
1
export MWS_TARGET_CLUSTER='<техническое имя целевого кластера>'
```
В примере техническое имя кластера — cluster-target-b, а не Target Kafka (отображаемое имя).

API Managed Kafka использует техническое имя кластера — последний сегмент из metadata.id. Если в запросе передать отображаемое имя, API вернет ошибку INVALID_ARGUMENT.

Проверьте доступ к API Managed Kafka:

1
curl -sS \
2
  -H "Authorization: Bearer ${TOKEN}" \
3
  "https://mkafka.mwsapis.ru/mkafka/v1/projects/${MWS_PROJECT}/clusters/${MWS_TARGET_CLUSTER}/topics" \
4
| jq

Ожидаемый результат: JSON-описание топиков целевого кластера.

4. Создайте топики в целевом кластере

MirrorMaker 2 в процессе работы попытается создать недостающие топики через Kafka Admin API. Пользователь кластера не имеет прав на создание топиков, поэтому такая операция завершится ошибкой TopicAuthorizationException.

Чтобы избежать этой ошибки, подготовьте топики целевого кластера одним из способов:

Скриптом
Через API

Скрипт sync-topics.sh создает в целевом кластере пользовательские топики исходного кластера, а также внутренние топики MM2, необходимые для миграции.

Используйте этот способ, чтобы перенести большое количество топиков.

Создайте на ВМ файл /opt/kafka-mm2/scripts/sync-topics.sh и вставьте в него содержимое скрипта:

1
#!/usr/bin/env bash
2
set -euo pipefail
3

4
API_BASE="https://mkafka.mwsapis.ru/mkafka/v1"
5
METADATA_TOKEN_URL="http://169.254.169.254/computeMetadata/v1/instance/service-accounts/default/token"
6
METADATA_SA_URL="http://169.254.169.254/computeMetadata/v1/instance/service-accounts/?recursive=true"
7

8
: "${MWS_PROJECT:?Set MWS_PROJECT}"
9
: "${MWS_TARGET_CLUSTER:?Set MWS_TARGET_CLUSTER}"
10
: "${SOURCE_BOOTSTRAP_SERVERS:?Set SOURCE_BOOTSTRAP_SERVERS}"
11

12
SOURCE_CLIENT_CONFIG="${SOURCE_CLIENT_CONFIG:-/etc/kafka-mm2/source-client.properties}"
13
KAFKA_TOPICS_BIN="${KAFKA_TOPICS_BIN:-/opt/kafka/current/bin/kafka-topics.sh}"
14

15
SOURCE_TOPICS_INCLUDE_REGEX="${SOURCE_TOPICS_INCLUDE_REGEX:-.*}"
16
SOURCE_TOPICS_EXCLUDE_REGEX="${SOURCE_TOPICS_EXCLUDE_REGEX:-^(__.*|mm2-.*[.]internal|.*[.]checkpoints[.]internal|heartbeats)$}"
17

18
SOURCE_ALIAS="${SOURCE_ALIAS:-source}"
19
REPLICATION_FACTOR="${REPLICATION_FACTOR:-1}"
20
CONNECT_OFFSET_PARTITIONS="${CONNECT_OFFSET_PARTITIONS:-1}"
21
CONNECT_STATUS_PARTITIONS="${CONNECT_STATUS_PARTITIONS:-1}"
22
WAIT_TIMEOUT_SECONDS="${WAIT_TIMEOUT_SECONDS:-300}"
23
WATCH_INTERVAL_SECONDS="${WATCH_INTERVAL_SECONDS:-0}"
24

25
for cmd in curl jq grep sed sort; do
26
  command -v "$cmd" >/dev/null 2>&1 || {
27
    echo "ERROR: required command '$cmd' is not installed." >&2
28
    exit 1
29
  }
30
done
31

32
[[ -x "$KAFKA_TOPICS_BIN" ]] || {
33
  echo "ERROR: kafka-topics.sh not found: $KAFKA_TOPICS_BIN" >&2
34
  exit 1
35
}
36

37
[[ -r "$SOURCE_CLIENT_CONFIG" ]] || {
38
  echo "ERROR: Source client config is not readable: $SOURCE_CLIENT_CONFIG" >&2
39
  echo "Run as kafka-mm2 or use SOURCE_CLIENT_CONFIG." >&2
40
  exit 1
41
}
42

43
TMP_RESPONSE="$(mktemp)"
44
TMP_METADATA="$(mktemp)"
45
trap 'rm -f "$TMP_RESPONSE" "$TMP_METADATA"' EXIT
46

47
check_metadata_service() {
48
  if ! curl -fsS --connect-timeout 2 --max-time 5       -H "Metadata-Flavor: Google"       "$METADATA_SA_URL"       -o "$TMP_METADATA"; then
49
    echo "ERROR: cannot access MWS VM metadata service." >&2
50
    echo "Attach a service account to this MWS VM and grant it access to manage Managed Kafka topics." >&2
51
    return 1
52
  fi
53

54
  if ! jq -e 'type == "object" and length > 0' "$TMP_METADATA" >/dev/null 2>&1; then
55
    echo "ERROR: no service account is visible in VM metadata." >&2
56
    cat "$TMP_METADATA" >&2
57
    return 1
58
  fi
59
}
60

61
get_mws_token() {
62
  local response token
63

64
  response="$(
65
    curl -fsS --connect-timeout 2 --max-time 10       -H "Metadata-Flavor: Google"       "$METADATA_TOKEN_URL"
66
  )" || {
67
    echo "ERROR: failed to obtain IAM token from MWS metadata service." >&2
68
    return 1
69
  }
70

71
  token="$(jq -er '.access_token | select(type=="string" and length>0)' <<<"$response")" || {
72
    echo "ERROR: metadata response does not contain access_token." >&2
73
    echo "$response" >&2
74
    return 1
75
  }
76

77
  printf '%s' "$token"
78
}
79

80
urlencode() {
81
  jq -rn --arg v "$1" '$v | @uri'
82
}
83

84
topic_url() {
85
  printf '%s/projects/%s/clusters/%s/topics/%s'     "$API_BASE"     "$(urlencode "$MWS_PROJECT")"     "$(urlencode "$MWS_TARGET_CLUSTER")"     "$(urlencode "$1")"
86
}
87

88
api_request() {
89
  local method="$1"
90
  local url="$2"
91
  local data="${3:-}"
92
  local token
93

94
  token="$(get_mws_token)"
95

96
  if [[ -n "$data" ]]; then
97
    curl -sS -o "$TMP_RESPONSE" -w "%{http_code}"       -X "$method" "$url"       -H "Authorization: Bearer ${token}"       -H "Content-Type: application/json"       -d "$data"
98
  else
99
    curl -sS -o "$TMP_RESPONSE" -w "%{http_code}"       -X "$method" "$url"       -H "Authorization: Bearer ${token}"
100
  fi
101
}
102

103
wait_topic_ready() {
104
  local topic="$1"
105
  local started="$SECONDS"
106

107
  while true; do
108
    local code state
109
    code="$(api_request GET "$(topic_url "$topic")")"
110

111
    if [[ "$code" != "200" ]]; then
112
      echo "ERROR: cannot read '$topic' while waiting (HTTP $code)." >&2
113
      cat "$TMP_RESPONSE" >&2
114
      return 1
115
    fi
116

117
    state="$(jq -r '.status.ready.state // empty' "$TMP_RESPONSE")"
118

119
    if [[ "$state" == "OK" ]]; then
120
      echo "  ready: $topic"
121
      return 0
122
    fi
123

124
    if (( SECONDS - started >= WAIT_TIMEOUT_SECONDS )); then
125
      echo "ERROR: timeout waiting for '$topic'; state='${state:-unknown}'." >&2
126
      cat "$TMP_RESPONSE" >&2
127
      return 1
128
    fi
129

130
    sleep 2
131
  done
132
}
133

134
create_or_validate_topic() {
135
  local topic="$1"
136
  local partitions="$2"
137
  local config="${3:-}"
138
  local code
139

140
  code="$(api_request GET "$(topic_url "$topic")")"
141

142
  if [[ "$code" == "200" ]]; then
143
    local actual_partitions actual_rf
144
    actual_partitions="$(jq -r '.spec.partitions // empty' "$TMP_RESPONSE")"
145
    actual_rf="$(jq -r '.spec.replicationFactor // empty' "$TMP_RESPONSE")"
146

147
    if [[ "$actual_partitions" != "$partitions" ]]; then
148
      echo "ERROR: '$topic' partition count differs." >&2
149
      echo "  expected/source: $partitions" >&2
150
      echo "  target:          ${actual_partitions:-unknown}" >&2
151
      return 1
152
    fi
153

154
    if [[ "$actual_rf" != "$REPLICATION_FACTOR" ]]; then
155
      echo "ERROR: '$topic' replication factor differs." >&2
156
      echo "  expected: $REPLICATION_FACTOR" >&2
157
      echo "  target:   ${actual_rf:-unknown}" >&2
158
      return 1
159
    fi
160

161
    echo "exists: $topic (partitions=$actual_partitions rf=$actual_rf)"
162
    wait_topic_ready "$topic"
163
    return 0
164
  fi
165

166
  if [[ "$code" != "404" ]]; then
167
    echo "ERROR: failed to check '$topic' (HTTP $code)." >&2
168
    cat "$TMP_RESPONSE" >&2
169
    return 1
170
  fi
171

172
  local payload
173
  payload="$(
174
    jq -n       --argjson partitions "$partitions"       --argjson rf "$REPLICATION_FACTOR"       --arg config "$config"       '{
175
        spec: {
176
          partitions: $partitions,
177
          replicationFactor: $rf,
178
          config: $config
179
        }
180
      } | if $config == "" then del(.spec.config) else . end'
181
  )"
182

183
  echo "create: $topic (partitions=$partitions rf=$REPLICATION_FACTOR${config:+ config='$config'})"
184

185
  code="$(api_request POST "$(topic_url "$topic")" "$payload")"
186

187
  if [[ "$code" != "200" ]]; then
188
    echo "ERROR: failed to create '$topic' (HTTP $code)." >&2
189
    cat "$TMP_RESPONSE" >&2
190
    return 1
191
  fi
192

193
  wait_topic_ready "$topic"
194
}
195

196
source_topics() {
197
  "$KAFKA_TOPICS_BIN"     --bootstrap-server "$SOURCE_BOOTSTRAP_SERVERS"     --command-config "$SOURCE_CLIENT_CONFIG"     --list   | grep -E "$SOURCE_TOPICS_INCLUDE_REGEX"   | grep -Ev "$SOURCE_TOPICS_EXCLUDE_REGEX"   | sort -u || true
198
}
199

200
source_partition_count() {
201
  local topic="$1"
202
  local description count
203

204
  description="$(
205
    "$KAFKA_TOPICS_BIN"       --bootstrap-server "$SOURCE_BOOTSTRAP_SERVERS"       --command-config "$SOURCE_CLIENT_CONFIG"       --describe       --topic "$topic"
206
  )"
207

208
  count="$(
209
    printf '%s\n' "$description"     | sed -n '1{s/.*PartitionCount:[[:space:]]*\([0-9][0-9]*\).*/\1/p;}'
210
  )"
211

212
  [[ -n "$count" ]] || {
213
    echo "ERROR: cannot determine PartitionCount for '$topic'." >&2
214
    echo "$description" >&2
215
    return 1
216
  }
217

218
  printf '%s' "$count"
219
}
220

221
create_internal_topics() {
222
  echo "== Kafka Connect internal topics =="
223
  create_or_validate_topic "mm2-configs.${SOURCE_ALIAS}.internal" 1 "cleanup.policy = compact"
224
  create_or_validate_topic "mm2-offsets.${SOURCE_ALIAS}.internal" "$CONNECT_OFFSET_PARTITIONS" "cleanup.policy = compact"
225
  create_or_validate_topic "mm2-status.${SOURCE_ALIAS}.internal" "$CONNECT_STATUS_PARTITIONS" "cleanup.policy = compact"
226

227
  echo
228
  echo "== MirrorMaker 2 internal topics =="
229
  create_or_validate_topic "mm2-offset-syncs.${SOURCE_ALIAS}.internal" 1 "cleanup.policy = compact"
230
  create_or_validate_topic "${SOURCE_ALIAS}.checkpoints.internal" 1 "cleanup.policy = compact"
231
  create_or_validate_topic "heartbeats" 1 "cleanup.policy = compact"
232
}
233

234
sync_user_topics() {
235
  echo
236
  echo "== Discover Source user topics =="
237
  echo "Source:  $SOURCE_BOOTSTRAP_SERVERS"
238
  echo "Include: $SOURCE_TOPICS_INCLUDE_REGEX"
239
  echo "Exclude: $SOURCE_TOPICS_EXCLUDE_REGEX"
240

241
  mapfile -t topics < <(source_topics)
242

243
  if (( ${#topics[@]} == 0 )); then
244
    echo "No Source user topics matched."
245
    return 0
246
  fi
247

248
  echo "Matched ${#topics[@]} topic(s):"
249
  printf '  %s\n' "${topics[@]}"
250

251
  echo
252
  echo "== Synchronize destination topics =="
253

254
  local topic partitions
255
  for topic in "${topics[@]}"; do
256
    partitions="$(source_partition_count "$topic")"
257
    echo "source: $topic partitions=$partitions"
258
    create_or_validate_topic "$topic" "$partitions"
259
  done
260
}
261

262
echo "Checking MWS metadata service..."
263
check_metadata_service
264
get_mws_token >/dev/null
265
echo "MWS service-account authentication: OK"
266
echo
267

268
echo "MWS project:        $MWS_PROJECT"
269
echo "Target cluster:     $MWS_TARGET_CLUSTER"
270
echo "MM2 source alias:   $SOURCE_ALIAS"
271
echo "Target RF:          $REPLICATION_FACTOR"
272
echo
273

274
create_internal_topics
275

276
while true; do
277
  sync_user_topics
278

279
  if (( WATCH_INTERVAL_SECONDS <= 0 )); then
280
    break
281
  fi
282

283
  echo
284
  echo "Next Source scan in ${WATCH_INTERVAL_SECONDS}s..."
285
  sleep "$WATCH_INTERVAL_SECONDS"
286
done
287

288
echo
289
echo "Topic synchronization completed."

Назначьте права на скрипт:

1
sudo chown kafka-mm2:kafka-mm2 \
2
  /opt/kafka-mm2/scripts/sync-topics.sh && \
3
sudo chmod 750 \
4
  /opt/kafka-mm2/scripts/sync-topics.sh

Выполните предварительную синхронизацию топиков:

1
sudo -u kafka-mm2 env \
2
  MWS_PROJECT='<имя проекта>' \
3
  MWS_TARGET_CLUSTER='<техническое имя целевого кластера>' \
4
  SOURCE_BOOTSTRAP_SERVERS='<адрес брокера исходного кластера>' \
5
  /opt/kafka-mm2/scripts/sync-topics.sh

Ожидаемый результат: скрипт создает топики и завершается сообщением Topic synchronization completed.

Проверьте топики целевого кластера:

1
sudo -u kafka-mm2 \
2
/opt/kafka/current/bin/kafka-topics.sh \
3
  --bootstrap-server <адрес брокера целевого кластера> \
4
  --command-config /etc/kafka-mm2/target-client.properties \
5
  --list

Ожидаемый результат: список содержит все пользовательские топики исходного кластера.

(опционально) Если во время миграции в исходном кластере появляются новые топики, запустите скрипт в режиме наблюдения. Скрипт будет периодически проверять исходный кластер и создавать новые топики в целевом:
bash
```
1
sudo -u kafka-mm2 env \
2
  MWS_PROJECT='<имя проекта>' \
3
  MWS_TARGET_CLUSTER='<техническое имя целевого кластера>' \
4
  SOURCE_BOOTSTRAP_SERVERS='<адрес брокера исходного кластера>' \
5
  WATCH_INTERVAL_SECONDS=30 \
6
  /opt/kafka-mm2/scripts/sync-topics.sh
```

Создайте топики по одному с помощью API Managed Kafka. Переменные ${MWS_PROJECT}, ${MWS_TARGET_CLUSTER} и ${TOKEN} уже заданы при проверке доступа к API.

Этот способ подходит для небольшого количества топиков.

Создайте пользовательский топик. Количество разделов укажите таким же, как в исходном кластере. В примере создается топик orders с шестью разделами:

1
TOPIC='orders' && \
2
PARTITIONS=6 && \
3
curl -sS \
4
  -X POST \
5
  "https://mkafka.mwsapis.ru/mkafka/v1/projects/${MWS_PROJECT}/clusters/${MWS_TARGET_CLUSTER}/topics/${TOPIC}" \
6
  -H "Authorization: Bearer ${TOKEN}" \
7
  -H "Content-Type: application/json" \
8
  -d "{
9
    \"spec\": {
10
      \"partitions\": ${PARTITIONS},
11
      \"replicationFactor\": 1
12
    }
13
  }" \
14
| jq

Создайте внутренние топики MM2. Количество разделов и фактор репликации для каждого топика — 1, конфигурация cleanup.policy всегда compact. Имена внутренних топиков:
- mm2-configs.source.internal
- mm2-offsets.source.internal
- mm2-status.source.internal
- mm2-offset-syncs.source.internal
- source.checkpoints.internal
Для кластера с несколькими брокерами выбирайте фактор репликации в соответствии с требованиями к отказоустойчивости.

Пример создания топика mm2-offsets.source.internal:
bash
```
1
TOPIC='mm2-offsets.source.internal' && \
2
curl -sS \
3
  -X POST \
4
  "https://mkafka.mwsapis.ru/mkafka/v1/projects/${MWS_PROJECT}/clusters/${MWS_TARGET_CLUSTER}/topics/${TOPIC}" \
5
  -H "Authorization: Bearer ${TOKEN}" \
6
  -H "Content-Type: application/json" \
7
  -d '{
8
    "spec": {
9
      "partitions": 1,
10
      "replicationFactor": 1,
11
      "config": "cleanup.policy = compact"
12
    }
13
  }' \
14
| jq
```
Повторите команду для каждого внутреннего топика.

Проверьте топики целевого кластера:

1
sudo -u kafka-mm2 \
2
/opt/kafka/current/bin/kafka-topics.sh \
3
  --bootstrap-server <адрес брокера целевого кластера> \
4
  --command-config /etc/kafka-mm2/target-client.properties \
5
  --list

Ожидаемый результат: список содержит все пользовательские топики исходного кластера.

5. Создайте конфигурацию MirrorMaker 2

Создайте файл /etc/kafka-mm2/connect-mirror-maker.properties со следующим содержимым:

1
########################################
2
# Кластеры
3
########################################
4

5
clusters = source, target
6

7
source.bootstrap.servers = <адрес брокера исходного кластера>
8
target.bootstrap.servers = <адрес брокера целевого кластера>
9

10

11
########################################
12
# Аутентификация в исходном кластере
13
########################################
14

15
source.security.protocol = SASL_SSL
16
source.sasl.mechanism = SCRAM-SHA-256
17

18
source.sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required username="<имя пользователя исходного кластера>" password="<пароль>";
19

20
source.ssl.truststore.type = PEM
21
source.ssl.truststore.location = /etc/kafka-mm2/mws-root-ca.crt
22
source.ssl.endpoint.identification.algorithm =
23

24

25
########################################
26
# Аутентификация в целевом кластере
27
########################################
28

29
target.security.protocol = SASL_SSL
30
target.sasl.mechanism = SCRAM-SHA-256
31

32
target.sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required username="<имя пользователя целевого кластера>" password="<пароль>";
33

34
target.ssl.truststore.type = PEM
35
target.ssl.truststore.location = /etc/kafka-mm2/mws-root-ca.crt
36
target.ssl.endpoint.identification.algorithm =
37

38

39
########################################
40
# Направление репликации
41
########################################
42

43
source->target.enabled = true
44
target->source.enabled = false
45

46

47
########################################
48
# Топики
49
########################################
50

51
source->target.topics = .*
52

53
source->target.topics.exclude = __.*,mm2-.*[.]internal,.*[.]checkpoints[.]internal,heartbeats
54

55
source->target.refresh.topics.enabled = true
56
source->target.refresh.topics.interval.seconds = 30
57

58

59
########################################
60
# Сохранение имен топиков
61
########################################
62

63
replication.policy.class = org.apache.kafka.connect.mirror.IdentityReplicationPolicy
64

65

66
########################################
67
# Настройки топиков
68
########################################
69

70
source->target.sync.topic.configs.enabled = false
71
source->target.sync.topic.acls.enabled = false
72

73
replication.factor = 1
74

75

76
########################################
77
# Синхронизация смещений
78
########################################
79

80
source->target.emit.offset-syncs.enabled = true
81
source->target.offset-syncs.topic.location = target
82

83
offset-syncs.topic.replication.factor = 1
84

85

86
########################################
87
# Контрольные точки
88
########################################
89

90
source->target.emit.checkpoints.enabled = true
91

92
checkpoints.topic.replication.factor = 1
93

94

95
########################################
96
# Heartbeats
97
########################################
98

99
source->target.emit.heartbeats.enabled = false
100
target->source.emit.heartbeats.enabled = false
101

102
source->target.heartbeats.replication.enabled = false
103

104
heartbeats.topic.replication.factor = 1
105

106

107
########################################
108
# Внутренние топики Kafka Connect
109
########################################
110

111
offset.storage.partitions = 1
112
status.storage.partitions = 1
113

114
offset.storage.replication.factor = 1
115
status.storage.replication.factor = 1
116
config.storage.replication.factor = 1
117

118

119
########################################
120
# Параллельность
121
########################################
122

123
source->target.tasks.max = 4

Значение .* для параметра source->target.topics включает все топики исходного кластера, а source->target.topics.exclude задает исключения и имеет приоритет над include-фильтром.

Задайте права на файл конфигурации:

1
sudo chown kafka-mm2:kafka-mm2 \
2
  /etc/kafka-mm2/connect-mirror-maker.properties && \
3
sudo chmod 600 \
4
  /etc/kafka-mm2/connect-mirror-maker.properties

В этой конфигурации заданы следующие ключевые решения:

Параметр	Значение	Причина
`replication.policy.class`	`IdentityReplicationPolicy`	Имена топиков переносятся без префикса исходного кластера
`source->target.emit.heartbeats.enabled`	`false`	Для миграции heartbeat-топики не нужны
`source->target.sync.topic.configs.enabled`	`false`	Топики целевого кластера создаются через API Managed Kafka
`replication.factor`	`1`	В примере кластер состоит из одного брокера
`source->target.tasks.max`	`4`	Количество параллельных задач репликации

Почему используется IdentityReplicationPolicy

По умолчанию MirrorMaker 2 добавляет к имени топика префикс исходного кластера: например, топик orders на целевом кластере изменит имя на source.orders. При миграции должны быть сохранены исходные имена топиков, поэтому применяется IdentityReplicationPolicy.

Почему отключены heartbeats

MirrorMaker 2 создает heartbeat-топики для отслеживания состояния репликации. В сочетании с IdentityReplicationPolicy он попытается создать топик source.heartbeats на целевом кластере и получит ошибку TopicAuthorizationException, т.к. MWS Cloud Platform не разрешает создание топиков через Kafka Admin API.

Для миграции данных heartbeat-механизм не обязателен, поэтому репликация heartbeat-топиков отключена параметрами source->target.emit.heartbeats.enabled, target->source.emit.heartbeats.enabled и source->target.heartbeats.replication.enabled.

Параметр heartbeats.replication.enabled по умолчанию равен true и заставляет MirrorSourceConnector реплицировать heartbeat-топик независимо от фильтра топиков.

Почему отключена синхронизация конфигураций и ACL

Топики целевого кластера создаются через API Managed Kafka, поэтому дополнительные административные операции через Kafka Admin API не нужны.

После проверки переноса данных сравните конфигурации топиков и перенесите необходимые настройки отдельно.

Почему replication.factor равен 1

В примере исходный и целевой кластеры состоят из одного брокера. Для кластера с несколькими брокерами задайте фактор репликации в соответствии с требованиями к отказоустойчивости.

Тот же принцип применяется к offset-syncs.topic.replication.factor, checkpoints.topic.replication.factor и параметрам хранилища Kafka Connect.

6. Проверьте направление репликации

Перед первым запуском репликации убедитесь, что адреса брокеров не перепутаны: данные должны идти из исходного кластера в целевой, а обратная репликация отключена.

Выполните команду:

1
grep -nE \
2
'^(clusters|source\.bootstrap\.servers|target\.bootstrap\.servers|source->target\.enabled|target->source\.enabled)' \
3
/etc/kafka-mm2/connect-mirror-maker.properties

Ожидаемый результат:

1
clusters = source, target
2

3
source.bootstrap.servers = <адрес брокера исходного кластера>
4
target.bootstrap.servers = <адрес брокера целевого кластера>
5

6
source->target.enabled = true
7
target->source.enabled = false

7. Запустите MirrorMaker 2

Запустите MirrorMaker 2:

1
sudo -u kafka-mm2 \
2
env KAFKA_HEAP_OPTS="-Xms512m -Xmx1g" \
3
/opt/kafka/current/bin/connect-mirror-maker.sh \
4
/etc/kafka-mm2/connect-mirror-maker.properties \
5
2>&1 | tee /tmp/mm2.log

В отдельном терминале проверьте работу процесса:

1
grep -E \
2
'MirrorSourceTask|replicating|ERROR|WARN' \
3
/tmp/mm2.log

Ожидаемый результат:

1
MirrorSourceTask
2
replicating 6 topic-partitions source->target:
3
[orders-0, orders-1, orders-2, orders-3, orders-4, orders-5]

8. Проверьте репликацию

Проверьте права записи в целевой кластер. Для этого отправьте тестовое сообщение:

1
echo '{"probe":"direct-target-write"}' | \
2
sudo -u kafka-mm2 \
3
/opt/kafka/current/bin/kafka-console-producer.sh \
4
  --bootstrap-server <адрес брокера целевого кластера> \
5
  --producer.config /etc/kafka-mm2/target-client.properties \
6
  --topic <имя топика>

Прочитайте его:

1
sudo -u kafka-mm2 \
2
/opt/kafka/current/bin/kafka-console-consumer.sh \
3
  --bootstrap-server <адрес брокера целевого кластера> \
4
  --consumer.config /etc/kafka-mm2/target-client.properties \
5
  --topic <имя топика> \
6
  --from-beginning \
7
  --max-messages 1

Ожидаемый результат: {"probe":"direct-target-write"}.

Проверьте перенос сообщений из исходного кластера:

1
sudo -u kafka-mm2 \
2
/opt/kafka/current/bin/kafka-console-consumer.sh \
3
  --bootstrap-server <адрес брокера целевого кластера> \
4
  --consumer.config /etc/kafka-mm2/target-client.properties \
5
  --topic <имя топика> \
6
  --from-beginning \
7
  --max-messages 5

Ожидаемый результат: потребитель выводит сообщения, отправленные в исходный кластер.

Проверьте прогресс миграции. Количество сообщений определяется разницей между последним и первым смещением: сообщения старше срока хранения удаляются из топика, поэтому одного последнего смещения недостаточно.

1
TOPIC='<имя топика>'
2

3
SOURCE_BOOTSTRAP='<адрес брокера исходного кластера>'
4
TARGET_BOOTSTRAP='<адрес брокера целевого кластера>'
5

6
SOURCE_CONFIG='/etc/kafka-mm2/source-client.properties'
7
TARGET_CONFIG='/etc/kafka-mm2/target-client.properties'
8

9

10
SOURCE_EARLIEST=$(
11
  sudo -u kafka-mm2 \
12
  /opt/kafka/current/bin/kafka-get-offsets.sh \
13
    --bootstrap-server "$SOURCE_BOOTSTRAP" \
14
    --command-config "$SOURCE_CONFIG" \
15
    --topic "$TOPIC" \
16
    --time -2 \
17
  | awk -F: '{sum += $3} END {print sum+0}'
18
)
19

20

21
SOURCE_LATEST=$(
22
  sudo -u kafka-mm2 \
23
  /opt/kafka/current/bin/kafka-get-offsets.sh \
24
    --bootstrap-server "$SOURCE_BOOTSTRAP" \
25
    --command-config "$SOURCE_CONFIG" \
26
    --topic "$TOPIC" \
27
    --time -1 \
28
  | awk -F: '{sum += $3} END {print sum+0}'
29
)
30

31

32
TARGET_EARLIEST=$(
33
  sudo -u kafka-mm2 \
34
  /opt/kafka/current/bin/kafka-get-offsets.sh \
35
    --bootstrap-server "$TARGET_BOOTSTRAP" \
36
    --command-config "$TARGET_CONFIG" \
37
    --topic "$TOPIC" \
38
    --time -2 \
39
  | awk -F: '{sum += $3} END {print sum+0}'
40
)
41

42

43
TARGET_LATEST=$(
44
  sudo -u kafka-mm2 \
45
  /opt/kafka/current/bin/kafka-get-offsets.sh \
46
    --bootstrap-server "$TARGET_BOOTSTRAP" \
47
    --command-config "$TARGET_CONFIG" \
48
    --topic "$TOPIC" \
49
    --time -1 \
50
  | awk -F: '{sum += $3} END {print sum+0}'
51
)
52

53

54
SOURCE_COUNT=$((SOURCE_LATEST - SOURCE_EARLIEST))
55
TARGET_COUNT=$((TARGET_LATEST - TARGET_EARLIEST))
56

57
LAG=$((SOURCE_COUNT - TARGET_COUNT))
58

59

60
if [ "$SOURCE_COUNT" -gt 0 ]; then
61
  PROGRESS=$(
62
    awk \
63
      -v target="$TARGET_COUNT" \
64
      -v source="$SOURCE_COUNT" \
65
      'BEGIN { printf "%.2f", (target / source) * 100 }'
66
  )
67
else
68
  PROGRESS="100.00"
69
fi
70

71

72
echo "Topic:     $TOPIC"
73
echo "Source:    $SOURCE_COUNT"
74
echo "Target:    $TARGET_COUNT"
75
echo "Lag:       $LAG"
76
echo "Progress:  ${PROGRESS}%"

Пример вывода скрипта подсчета:

1
Topic:     mm2.test.orders
2
Source:    1000000
3
Target:    734521
4
Lag:       265479
5
Progress:  73.45%

Остановите производителей в исходном кластере и дождитесь значения Progress: 100.00% — это будет означать, что миграция успешно завершена.

Пока производители отправляют сообщения, значение прогресса будет постоянно изменяться.

Типовые ошибки

SSL handshake: PKIX path building failed

1
SSL handshake failed
2

3
PKIX path building failed
4

5
unable to find valid certification path

Проверьте, что существует файл сертификата MWS /etc/kafka-mm2/mws-root-ca.crt, а в файлах конфигурации MM2 указаны параметры ssl.truststore.type=PEM и ssl.truststore.location=/etc/kafka-mm2/mws-root-ca.crt.

SASL authentication failed

Проверьте параметры аутентификации в файле конфигурации MM2: security.protocol=SASL_SSL, sasl.mechanism=SCRAM-SHA-256 и учетные данные в sasl.jaas.config.

Could not look up partition metadata

1
ConnectException:
2

3
Could not look up partition metadata for topic
4
'mm2-offsets.source.internal'

MM2 попытался создать внутренний топик через Kafka Admin API. Чтобы устранить ошибку, создайте топики в целевом кластере.

TopicAuthorizationException при создании топика

1
Could not create topic ...
2

3
TopicAuthorizationException:
4

5
Authorization failed

Создание топиков в Managed Kafka выполняется через API сервиса, а не через Kafka Admin API. Роль CLUSTER_ADMIN этого не меняет. Предварительно создайте топики в целевом кластере.

Could not create topic source.heartbeats

1
Could not create topic source.heartbeats

Отключите репликацию heartbeat-топиков. Для этого используйте корректную конфигурацию.

INVALID_ARGUMENT listKafkaTopics.cluster

1
{
2
  "code": "INVALID_ARGUMENT",
3
  "description": "listKafkaTopics.cluster has invalid value"
4
}

В запрос передано отображаемое имя кластера вместо технического. Используйте техническое имя кластера.

Нет данных в целевом кластере

Сначала проверьте имя топика. При IdentityReplicationPolicy имя топика не меняется: если в исходном кластере топик называется mm2.test.orders, то и в целевом он должен называться mm2.test.orders (но не orders).

Затем проверьте лог MirrorMaker 2:

1
grep -E \
2
'MirrorSourceTask|replicating|ERROR|WARN' \
3
/tmp/mm2.log

Удалите платные ресурсы

Ресурсы, созданные в руководстве, тарифицируются. Если вы больше не планируете использовать их:

Удалите виртуальную машину.
Удалите сервисный аккаунт.
Удалите целевой кластер Managed Kafka, если вы проходили руководство на тестовых данных.