Cp-docker-images: Setting custom Configurations for Producer & Consumer via Kafka Connect running in Distributed Mode

[Arun-Trichy]

I have a requirement to send Large Messages greater than 1 MB through Kafka. Default Properties to be changed in this case includes
Producer: max.request.size
Consumer: max.partition.fetch.bytes
Topic / Broker: max.message.bytes & replica.fetch.max.bytes

Referred from: https://stackoverflow.com/a/39026744/5758203

Command to create Kafka Topic that accepts > 1 MB

./kafka-topics --create --zookeeper zookeeper:2181 --topic Mytopic --partitions 1 --replication-factor 1 --config max.message.bytes=5048576

I have written custom Kafka Source Connectors to pull data from S3 and put it into a topic. I tried running it in distributed mode setting the properties "producer.max.request.size". I had various approaches:

• ## Set "producer.max.request.size" in the CuRL request
  > curl -X POST -H "Content-Type: application/json" --data '{"name":"auto_dataset_source","config":{"connector.class":"connectors.source.s3.S3Connector","tasks.max":"1","topic":"Mytopic","bucket":"kafkaS3Bucket","s3.endpoint":"s3-ap-southeast-2.amazonaws.com","s3.accesskey":"$$$$$$$$$$$$$","s3.secretkey":"$$$$$$$$$$$$$$$$$$$$$$$$$","key.converter":"org.apache.kafka.connect.json.JsonConverter","value.converter":"kafkautils.commons.Convertors.KafkaTensorProtoConvertor","producer.max.request.size":"5048576"}}' http://localhost:8083/connectors | jq

But the value was not getting set, and I keep getting the error:

[2018-03-19 08:57:06,057] ERROR auto_dataset_source_001-0 failed to send record to Mytopic: {} (org.apache.kafka.connect.runtime.WorkerSourceTask)
org.apache.kafka.common.errors.RecordTooLargeException: The message is 1311767 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.

• ## Set "producer.max.request.size" while starting the connect cluster in the compose file
  > connect:
  image: ai/cp-kafka-connect
  hostname: connect
  depends_on:
  
  
  
  • zookeeper
  
  
  • broker
    
    ports:
  
  
  • "8083:8083"
    
    environment:
    
    CONNECT_BOOTSTRAP_SERVERS: 'broker:9092'
    
    CONNECT_REST_ADVERTISED_HOST_NAME: connect
    
    CONNECT_REST_PORT: 8083
    
    CONNECT_GROUP_ID: compose-connect-group
    
    CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
    
    CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
    
    CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000
    
    CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
    
    CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
    
    CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
    
    CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
    
    CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
    
    CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
    
    CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
    
    CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
    
    CONNECT_ZOOKEEPER_CONNECT: 'zookeeper:2181'
    
    CONNECT_MAX_REQUEST_SIZE: 5048576
  
  

The value got set in the kafka connect cluster configuration. But when I started the connector using CuRL it started taking the default value for the Producer & Consumer and threw the same error

[2018-03-19 07:23:09,990] INFO Kafka version : 0.11.0.1-cp1 (org.apache.kafka.common.utils.AppInfoParser)
[2018-03-19 07:23:09,991] INFO Kafka commitId : 58558be45f16e1e4 (org.apache.kafka.common.utils.AppInfoParser)
[2018-03-19 07:23:10,111] INFO jetty-9.2.22.v20170606 (org.eclipse.jetty.server.Server)
[2018-03-19 07:23:11,115] INFO Started o.e.j.s.ServletContextHandler@3b6b1c41{/,null,AVAILABLE} (org.eclipse.jetty.server.handler.ContextHandler)
[2018-03-19 07:23:11,127] INFO Started ServerConnector@684e30c9{HTTP/1.1}{0.0.0.0:8083} (org.eclipse.jetty.server.ServerConnector)
[2018-03-19 07:23:11,127] INFO Started @14787ms (org.eclipse.jetty.server.Server)
[2018-03-19 07:23:11,130] INFO Kafka Connect started (org.apache.kafka.connect.runtime.Connect)
[2018-03-19 07:23:11,482] INFO Created topic (name=docker-connect-offsets, numPartitions=25, replicationFactor=1, replicasAssignments=null, configs={cleanup.policy=compact}) on brokers at broker:9092 (org.apache.kafka.connect.util.TopicAdmin)
[2018-03-19 07:23:11,498] INFO ProducerConfig values:
acks = all
batch.size = 16384
bootstrap.servers = [broker:9092]
buffer.memory = 33554432
client.id =
compression.type = none
confluent.batch.expiry.ms = 30000
connections.max.idle.ms = 540000
enable.idempotence = false
interceptor.classes = null
key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
linger.ms = 0
max.block.ms = 60000
max.in.flight.requests.per.connection = 1
max.request.size = 5048576
....
[2018-03-19 07:39:18,837] INFO Instantiated task auto_dataset_source-0 with version 0.0.1 of type connectors.source.s3.S3Task (org.apache.kafka.connect.runtime.Worker)
[2018-03-19 07:39:18,839] INFO ProducerConfig values:
acks = all
batch.size = 16384
bootstrap.servers = [broker:9092]
buffer.memory = 33554432
client.id =
compression.type = none
confluent.batch.expiry.ms = 30000
connections.max.idle.ms = 540000
enable.idempotence = false
interceptor.classes = null
key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
linger.ms = 0
max.block.ms = 9223372036854775807
max.in.flight.requests.per.connection = 1
max.request.size = 1048576
....
[2018-03-19 07:39:49,505] ERROR auto_dataset_source-0 failed to send record to Mytopic: {} (org.apache.kafka.connect.runtime.WorkerSourceTask)
org.apache.kafka.common.errors.RecordTooLargeException: The message is 1311767 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.

Which is the right place to update Producer/ Consumer specific configurations for the custom kafka connectors running in distributed mode?

[Arun-Trichy]

I read the documentation given here.

When you say, "For configuration of Kafka source and Kafka sink tasks, the same parameters can be used but need to be prefixed with consumer. and producer. respectively."

• Which all parameters do you refer to? Do you refer to the parameters used for running a Producer & Consumer API ?

When you say, "These parameters will need to be set up to three times in the worker configuration, once for management access, once for Kafka sinks and once for Kafka sources."

• What do you mean by it?

When you say, " ...the default configuration provided by config/server.properties."

• What all configurations do you refer to?

[Arun-Trichy]

Finally, got it solved. This link helped me get the issue.

I have to set all the below properties while spinning up the clusters of KAFKA & KAFKA-CONNECT via compose file.

KAFKA_REPLICA_FETCH_MAX_BYTES: 5048576
KAFKA_PRODUCER_MAX_REQUEST_SIZE: 5048576
KAFKA_CONSUMER_MAX_PARTITION_FETCH_BYTES: 5048576

CONNECT_PRODUCER_MAX_REQUEST_SIZE: 5048576
CONNECT_CONSUMER_MAX_PARTITION_FETCH_BYTES: 5048576

And then, the curl request for Source & Sink connector should be

curl -X POST -H "Content-Type: application/json" --data '{"name":"auto_dataset_source","config":{"connector.class":"connectors.source.s3.S3Connector","tasks.max":"1","topic":"Mytopic","bucket":"kafkaS3Bucket","s3.endpoint":"s3-ap-southeast-2.amazonaws.com","s3.accesskey":"$$$$$$$$$$$$$","s3.secretkey":"$$$$$$$$$$$$$$$$$$$$$$$$$","key.converter":"org.apache.kafka.connect.json.JsonConverter","value.converter":"kafkautils.commons.Convertors.KafkaTensorProtoConvertor","max.request.size":"5048576"}}' http://localhost:8083/connectors | jq

curl -X POST -H "Content-Type: application/json" --data '{"name":"auto_dataset_sink","config":{"connector.class":"connectors.sink.s3.S3SinkConnector","s3.endpoint":"s3-ap-southeast-2.amazonaws.com","s3.accesskey":"$$$$$$$$$$$$$","s3.secretkey":"$$$$$$$$$$$$$$$$$$$$$$$$$","tasks.max":"1","topics":"Mytopic","s3.bucketname":"kafkaS3BucketTarget ","max.partition.fetch.bytes":"5048576"}}' http://localhost:8083/connectors | jq

[b-luu]

_Putting this here for reference as it's the most relevant ticket I've found. (even if it concerns usage via docker...)_

If you're using the connect.properties file (either in standalone or distributed mode) the needed parameter to add to that file is essentially: producer.max.request.size=16777216 (for 16MiB for example... The default being 1MiB).

Notes:

• The .*fetch.*bytes properties are not Producer properties but Broker (replication) and Consumer ones. And they're not hard limits (request with bigger size than the set maximum will still be processed...)
• max.request.size on the connect.properties file is taken into account for the producer that writes to the connect-configs, connect-status and connect-offsets topics which handle the metadata of the connectors launched on the connect instance(s)
• Therefor, it's actually another totally separate producer with its own configuration (that can be modified with the producer. prefix in the connect.properties file as highlighted at the top of this comment) that's added for source connectors (cf https://docs.confluent.io/3.2.0/connect/userguide.html#overriding-producer-consumer-settings).

[mananstar2001]

We are trying to configure consumer group id at kafka connect. Our kafka cluster is ACL driven which allows, certain naming convention for consumers, which need to be changed from default pattern "connect-". Please guide in getting the prefix pattern changed.