Я пытаюсь перенести тематические данные из одного кластера Kafka в другой (оба на-linux).Для этого я использую Confluent kafka-connect-replicator.
Using - confluent enterprise 5.0
Следуя этому документу - https://docs.confluent.io/current/connect/kafka-connect-replicator/index.html#replicator-as-a-connector
Я уже изменил порты zookeeperи брокер до 2171 и 9082 соотв.у источника кафка кластера.Целевой кластер Kafka работает на портах по умолчанию.
Мои файлы конфигурации:
/etc/kafka-connect-replicator/quickstart-replicator.properties
name=replicator-source
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=X.X.X.X:9082
src.zookeeper.connect=X.X.X.X:2171
dest.kafka.bootstrap.servers=Y.Y.Y.Y:9092
dest.zookeeper.connect=Y.Y.Y.Y:2181
confluent.topic.replication.factor=1
topic.whitelist=test-topic
topic.rename.format=${topic}.replica
topic.create.backoff.ms=10000
/ etc / schema-registry / connect-avro-standalone.properties
bootstrap.servers=localhost:9082
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
offset.storage.file.filename=/data/kafka/connect.offsets
plugin.path=/usr/share/java
Запустите соединитель репликатора -
/usr/bin/connect-standalone /etc/schema-registry/connect-avro-standalone.properties /etc/kafka-connect-replicator/quickstart-replicator.properties
Но когда я запускаю репликатор, он создает тему реплики в целевом кластере, но если я создаю сообщения, он не отображается в приемнике в целевом кластере.В командной строке также выдаются предупреждения.
Nov 28, 2018 2:22:26 PM org.glassfish.jersey.internal.inject.Providers checkProviderRuntime
WARNING: A provider org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource registered in SERVER runtime does not implement any provider interfaces applicable in the SERVER runtime. Due to constraint configuration problems the provider org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource will be ignored.
Nov 28, 2018 2:22:26 PM org.glassfish.jersey.internal.inject.Providers checkProviderRuntime
WARNING: A provider org.apache.kafka.connect.runtime.rest.resources.RootResource registered in SERVER runtime does not implement any provider interfaces applicable in the SERVER runtime. Due to constraint configuration problems the provider org.apache.kafka.connect.runtime.rest.resources.RootResource will be ignored.
Nov 28, 2018 2:22:26 PM org.glassfish.jersey.internal.inject.Providers checkProviderRuntime
WARNING: A provider org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource registered in SERVER runtime does not implement any provider interfaces applicable in the SERVER runtime. Due to constraint configuration problems the provider org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource will be ignored.
Nov 28, 2018 2:22:26 PM org.glassfish.jersey.internal.Errors logErrors
WARNING: The following warnings have been detected: WARNING: The (sub)resource method createConnector in org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource contains empty path annotation.
WARNING: The (sub)resource method listConnectors in org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource contains empty path annotation.
WARNING: The (sub)resource method listConnectorPlugins in org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource contains empty path annotation.
WARNING: The (sub)resource method serverInfo in org.apache.kafka.connect.runtime.rest.resources.RootResource contains empty path annotation.
connect.log
[2018-11-28 14:29:58,849] INFO WorkerSourceTask{id=replicator-source-3} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:397)
[2018-11-28 14:29:58,849] INFO WorkerSourceTask{id=replicator-source-3} flushing 1 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:414)
[2018-11-28 14:30:03,850] ERROR WorkerSourceTask{id=replicator-source-3} Failed to flush, timed out while waiting for producer to flush outstanding 1 messages (org.apache.kafka.connect.runtime.WorkerSourceTask:419)
[2018-11-28 14:30:03,850] ERROR WorkerSourceTask{id=replicator-source-3} Failed to commit offsets (org.apache.kafka.connect.runtime.SourceTaskOffsetCommitter:111)
Чего мне не хватает?Как я могу реплицировать данные другому брокеру кластера?
спасибо!