[
https://issues.apache.org/jira/browse/KAFKA-10413?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17860514#comment-17860514
]
yazgoo edited comment on KAFKA-10413 at 6/27/24 2:47 PM:
---------------------------------------------------------
Here is yet another simpler version of the script, with less workers and which
does not try and restart any worker:
{code:java}
#!/bin/bash
set -xe
dkill() {
docker stop "$1" || true
docker rm -v -f "$1" || true
}
write_topic() {
# write 200 messages to the topic
json='{"name": "test"}'
docker exec -i kafka bash -c "(for i in {1..200}; do echo '$json'; done) |
/opt/kafka/bin/kafka-console-producer.sh --bootstrap-server 0.0.0.0:9092
--topic test_topic$1"
}
launch_minio() {
# Launch Minio (Fake S3)
docker run --network host -d --name minio \
-e MINIO_ROOT_USER=minioadmin \
-e MINIO_ROOT_PASSWORD=minioadmin \
minio/minio server --console-address :9001 /data
docker exec -it minio mkdir /data/my-minio-bucket
}
launch_kafka_connect() {
# Start Kafka Connect with S3 Connector
docker run --network host -d --name "kafka-connect$1" \
-e AWS_ACCESS_KEY_ID=minioadmin \
-e AWS_SECRET_ACCESS_KEY=minioadmin \
-e CONNECT_REST_ADVERTISED_HOST_NAME="k$1" \
-e CONNECT_LISTENERS="http://localhost:808$1"; \
-e CONNECT_BOOTSTRAP_SERVERS=0.0.0.0:9092 \
-e CONNECT_REST_PORT="808$1" \
-e CONNECT_GROUP_ID="connect-cluster" \
-e CONNECT_CONFIG_STORAGE_TOPIC="connect-configs" \
-e CONNECT_OFFSET_STORAGE_TOPIC="connect-offsets" \
-e CONNECT_STATUS_STORAGE_TOPIC="connect-status" \
-e CONNECT_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
-e CONNECT_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
-e CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE=false \
-e
CONNECT_INTERNAL_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
-e
CONNECT_INTERNAL_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
-e CONNECT_INTERNAL_VALUE_CONVERTER_SCHEMAS_ENABLE=false \
-e
CONNECT_PLUGIN_PATH="/usr/share/java,/usr/share/confluent-hub-components" \
--entrypoint bash \
confluentinc/cp-kafka-connect:7.6.1 \
-c "confluent-hub install --no-prompt confluentinc/kafka-connect-s3:latest
&& /etc/confluent/docker/run"
}
cleanup_docker_env() {
docker volume prune -f
for container in $(for i in {1..9}; do echo "kafka-connect$i";done) kafka
minio
do
dkill "$container"
done
}
launch_kafka() {
docker run --network host --hostname localhost --ulimit nofile=65536:65536 -d
--name kafka -p 9092:9092 apache/kafka
for i in {1..2}
do
# Create a Kafka topic
docker exec -it kafka /opt/kafka/bin/kafka-topics.sh --create
--bootstrap-server 0.0.0.0:9092 --replication-factor 1 --partitions 120 --topic
"test_topic$i"
write_topic "$i"
done
for topic in connect-configs connect-offsets connect-status
do
# with cleanup.policy=compact, we can't have more than 1 partition
docker exec -it kafka /opt/kafka/bin/kafka-topics.sh --create
--bootstrap-server 0.0.0.0:9092 --replication-factor 1 --partitions 1 --topic
$topic --config cleanup.policy=compact
done
}
cleanup_docker_env
launch_kafka
launch_minio
launch_kafka_connect 1
while true
do
sleep 5
# Check if Kafka Connect is up
curl http://localhost:8081/ || continue
break
done
sleep 10
for i in {1..2}
do
# Set up a connector
curl -X POST -H "Content-Type: application/json" --data '{
"name": "s3-connector'"$i"'",
"config": {
"connector.class": "io.confluent.connect.s3.S3SinkConnector",
"tasks.max": "120",
"topics": "test_topic'"$i"'",
"s3.region": "us-east-1",
"store.url": "http://0.0.0.0:9000";,
"s3.bucket.name": "my-minio-bucket",
"s3.part.size": "5242880",
"flush.size": "3",
"storage.class": "io.confluent.connect.s3.storage.S3Storage",
"format.class": "io.confluent.connect.s3.format.json.JsonFormat",
"schema.generator.class":
"io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator",
"schema.compatibility": "NONE"
}
}' http://localhost:8081/connectors
done
launch_kafka_connect 2
launch_kafka_connect 3
{code}
When the scrit ends, I have two each connector one worker with connector #1
tasks, the other one with connector #2 tasks.
Then I wait 3 minutes
And I get the final state:
{code:java}
❯ curl -s http://localhost:8081/connectors/s3-connector2/status | jq .tasks
|grep worker_id | sort | uniq -c
60 "worker_id": "k2:8082"
60 "worker_id": "k3:8083"{code}
{code:java}
❯ curl -s http://localhost:8081/connectors/s3-connector1/status | jq .tasks
|grep worker_id | sort | uniq -c
80 "worker_id": "k1:8081"
20 "worker_id": "k2:8082"
20 "worker_id": "k3:8083"
{code}
In the end, we indeed get 80 tasks on each workers, but for distribution
reasons , I think it should be (40, 40, 40) for each connector
was (Author: yazgoo):
Here is yet another simpler version of the script, with less workers and which
does not try and restart any worker:
{code:java}
{code}
{code:java}
#!/bin/bash
set -xe
dkill() {
docker stop "$1" || true
docker rm -v -f "$1" || true
}
write_topic() {
# write 200 messages to the topic
json='{"name": "test"}'
docker exec -i kafka bash -c "(for i in {1..200}; do echo '$json'; done) |
/opt/kafka/bin/kafka-console-producer.sh --bootstrap-server 0.0.0.0:9092
--topic test_topic$1"
}
launch_minio() {
# Launch Minio (Fake S3)
docker run --network host -d --name minio \
-e MINIO_ROOT_USER=minioadmin \
-e MINIO_ROOT_PASSWORD=minioadmin \
minio/minio server --console-address :9001 /data
docker exec -it minio mkdir /data/my-minio-bucket
}
launch_kafka_connect() {
# Start Kafka Connect with S3 Connector
docker run --network host -d --name "kafka-connect$1" \
-e AWS_ACCESS_KEY_ID=minioadmin \
-e AWS_SECRET_ACCESS_KEY=minioadmin \
-e CONNECT_REST_ADVERTISED_HOST_NAME="k$1" \
-e CONNECT_LISTENERS="http://localhost:808$1"; \
-e CONNECT_BOOTSTRAP_SERVERS=0.0.0.0:9092 \
-e CONNECT_REST_PORT="808$1" \
-e CONNECT_GROUP_ID="connect-cluster" \
-e CONNECT_CONFIG_STORAGE_TOPIC="connect-configs" \
-e CONNECT_OFFSET_STORAGE_TOPIC="connect-offsets" \
-e CONNECT_STATUS_STORAGE_TOPIC="connect-status" \
-e CONNECT_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
-e CONNECT_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
-e CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE=false \
-e
CONNECT_INTERNAL_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
-e
CONNECT_INTERNAL_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
-e CONNECT_INTERNAL_VALUE_CONVERTER_SCHEMAS_ENABLE=false \
-e
CONNECT_PLUGIN_PATH="/usr/share/java,/usr/share/confluent-hub-components" \
--entrypoint bash \
confluentinc/cp-kafka-connect:7.6.1 \
-c "confluent-hub install --no-prompt confluentinc/kafka-connect-s3:latest
&& /etc/confluent/docker/run"
}
cleanup_docker_env() {
docker volume prune -f
for container in $(for i in {1..9}; do echo "kafka-connect$i";done) kafka
minio
do
dkill "$container"
done
}
launch_kafka() {
docker run --network host --hostname localhost --ulimit nofile=65536:65536 -d
--name kafka -p 9092:9092 apache/kafka
for i in {1..2}
do
# Create a Kafka topic
docker exec -it kafka /opt/kafka/bin/kafka-topics.sh --create
--bootstrap-server 0.0.0.0:9092 --replication-factor 1 --partitions 120 --topic
"test_topic$i"
write_topic "$i"
done
for topic in connect-configs connect-offsets connect-status
do
# with cleanup.policy=compact, we can't have more than 1 partition
docker exec -it kafka /opt/kafka/bin/kafka-topics.sh --create
--bootstrap-server 0.0.0.0:9092 --replication-factor 1 --partitions 1 --topic
$topic --config cleanup.policy=compact
done
}
cleanup_docker_env
launch_kafka
launch_minio
launch_kafka_connect 1
while true
do
sleep 5
# Check if Kafka Connect is up
curl http://localhost:8081/ || continue
break
done
sleep 10
for i in {1..2}
do
# Set up a connector
curl -X POST -H "Content-Type: application/json" --data '{
"name": "s3-connector'"$i"'",
"config": {
"connector.class": "io.confluent.connect.s3.S3SinkConnector",
"tasks.max": "120",
"topics": "test_topic'"$i"'",
"s3.region": "us-east-1",
"store.url": "http://0.0.0.0:9000";,
"s3.bucket.name": "my-minio-bucket",
"s3.part.size": "5242880",
"flush.size": "3",
"storage.class": "io.confluent.connect.s3.storage.S3Storage",
"format.class": "io.confluent.connect.s3.format.json.JsonFormat",
"schema.generator.class":
"io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator",
"schema.compatibility": "NONE"
}
}' http://localhost:8081/connectors
done
launch_kafka_connect 2
launch_kafka_connect 3
{code}
When the scrit ends, I have two each connector one worker with connector #1
tasks, the other one with connector #2 tasks.
Then I wait 3 minutes
And I get the final state:
{code:java}
❯ curl -s http://localhost:8081/connectors/s3-connector2/status | jq .tasks
|grep worker_id | sort | uniq -c
60 "worker_id": "k2:8082"
60 "worker_id": "k3:8083"{code}
{code:java}
❯ curl -s http://localhost:8081/connectors/s3-connector1/status | jq .tasks
|grep worker_id | sort | uniq -c
80 "worker_id": "k1:8081"
20 "worker_id": "k2:8082"
20 "worker_id": "k3:8083"
{code}
In the end, we indeed get 80 tasks on each workers, but for distribution
reasons , I think it should be (40, 40, 40) for each connector
> rebalancing leads to unevenly balanced connectors
> -------------------------------------------------
>
> Key: KAFKA-10413
> URL: https://issues.apache.org/jira/browse/KAFKA-10413
> Project: Kafka
> Issue Type: Bug
> Components: connect
> Affects Versions: 2.5.1
> Reporter: yazgoo
> Assignee: rameshkrishnan muthusamy
> Priority: Major
> Fix For: 2.4.2, 2.5.2, 2.8.0, 2.7.1, 2.6.2
>
> Attachments: connect_worker_balanced.png, rebalance.sh
>
>
> GHi,
> With CP 5.5, running kafka connect s3 sink on EC2 whith autoscaling enabled,
> if a connect instance disappear, or a new one appear, we're seeing unbalanced
> consumption, much like mentionned in this post:
> [https://stackoverflow.com/questions/58644622/incremental-cooperative-rebalancing-leads-to-unevenly-balanced-connectors]
> This usually leads to one kafka connect instance taking most of the load and
> consumption not being able to keep on.
> Currently, we're "fixing" this by deleting the connector and re-creating it,
> but this is far from ideal.
> Any suggestion on what we could do to mitigate this ?
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
