Я скачал конфлюэнтную версию 5.4.
И у меня исправно работает разъем в автономном режиме.
wget /usr/local http://packages.confluent.io/archive/5.4/confluent-5.4.0-2.11.tar.gz -P ~/Downloads/
tar -zxvf ~/Downloads/confluent-5.4.0-2.11.tar.gz -C ~/Downloads/
sudo mv ~/Downloads/confluent-5.4.0 /usr/local/
, а затем я пытаюсь запустить Kafak подключиться в распределенном режиме, поэтому я изменил свои connect-distributed.properties, как показано ниже
bootstrap.servers=b-***.eu-west-1.amazonaws.com:9092,b-***.eu-west-1.amazonaws.com:9092,b-***.eu-west-1.amazonaws.com:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
config.storage.topic=connect-configs
config.storage.replication.factor=1
status.storage.topic=connect-status
status.storage.replication.factor=1
offset.flush.interval.ms=10000
plugin.path=/usr/local/confluent/share/java
, а затем я запускаю свой разъем, как показано ниже
/usr/local/confluent/bin/connect-distributed /usr/local/confluent/etc/kafka/connect-distributed.properties
, похоже, он запустился успешно [2020- 02-02 05: 22: 33,860] INFO Присоединился к группе и получил назначение:
Assignment{error=0, leader='connect-1-c99d50a9-faf0-4b15-8a3d-3add55b7e206', leaderUrl='http://10.97.49.217:8083/', offset=-1, connectorIds=[], taskIds=[]} (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1216)
[2020-02-02 05:22:33,861] INFO Starting connectors and tasks using config offset -1 (org.apache.kafka.connect.runtime.distributed.DistributedHerder:850)
[2020-02-02 05:22:33,861] INFO Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder:860)
Наконец, я использую команду curl для создания моего коннектора раковины.
curl -X POST -H "Content-Type: application/json" --data '{"name":"elastic-search-sink-audit","config":{"connector.class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector","tasks.max":"2","topics":"fsp-AUDIT_EVENT_DEMO","key.ignore":"true","connection.url":"https://**.amazonaws.com","type.name":"kafka-connect","name":"elastic-search-sink-audit","errors.tolerance":"all","errors.deadletterqueue.topic.name":"fsp-dlq-audit-event"}}' http://localhost:8083/connectors | jq
и с EC2 IP адрес
curl -X POST -H "Content-Type: application/json" --data '{"name":"elastic-search-sink-audit-distributed","config":{"connector.class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector","tasks.max":"2","topics":"audit_event_distributed","key.ignore":"true","connection.url":"https://***.us-east-1.es.amazonaws.com","type.name":"kafka-connect","name":"elastic-search-sink-audit-distributed","errors.tolerance":"all","errors.deadletterqueue.topic.name":"dlq_distributed"}}' http://10.67.39.217:8083/connectors | jq
После выполнения этой команды через 1 минуту я получаю ответ ниже
% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
100 498 0 48 100 450 0 4 0:01:52 0:01:30 0:00:22 10
{
"error_code": 500,
"message": "Request timed out"
}
И затем я продолжаю получать это, но это просто предупреждение, так как у меня достаточно реплик в моем топи c
NOT_ENOUGH_REPLICAS (org.apache.kafka.clients.producer.internals.Sender:598)
^C[2020-02-03 08:05:54,890] WARN [Producer clientId=producer-3] Got error produce response with correlation id 7185 on topic-partition connect-configs-0, retrying (2147476495 attempts left). Error: NOT_ENOUGH_REPLICAS (org.apache.kafka.clients.producer.internals.Sender:598)
^C[2020-02-03 08:05:54,991] WARN [Producer clientId=producer-3] Got error produce response with correlation id 7186 on topic-partition connect-configs-0, retrying (2147476494 attempts left). Error: NOT_ENOUGH_REPLICAS (org.apache.kafka.clients.producer.internals.Sender:598)
Когда я описал мой топи c
Topic:fsp-AUDIT_EVENT_DEMO PartitionCount:1 ReplicationFactor:3 Configs:
Topic: fsp-AUDIT_EVENT_DEMO Partition: 0 Leader: 1 Replicas: 1,6,2 Isr: 1,6,2
Так что у меня достаточно реплик для моего топи c.
ОБНОВЛЕНИЕ
Опишите результат
Topic:connect-configs PartitionCount:1 ReplicationFactor:1 Configs:cleanup.policy=compact
Topic: connect-configs Partition: 0 Leader: 6 Replicas: 6 Isr: 6
Topic:connect-status PartitionCount:5 ReplicationFactor:1 Configs:cleanup.policy=compact
Topic: connect-status Partition: 0 Leader: 6 Replicas: 6 Isr: 6
Topic: connect-status Partition: 1 Leader: 1 Replicas: 1 Isr: 1
Topic: connect-status Partition: 2 Leader: 2 Replicas: 2 Isr: 2
Topic: connect-status Partition: 3 Leader: 4 Replicas: 4 Isr: 4
Topic: connect-status Partition: 4 Leader: 5 Replicas: 5 Isr: 5
Topic:connect-offsets PartitionCount:25 ReplicationFactor:1 Configs:cleanup.policy=compact
Пожалуйста, помогите я разрешаю это.