Overriding Consumer, Producer Configuration via Env Variables

[spruisken]

I have a docker-compose file with Zookeeper, Kafka, Schema Registry, and Kafka Connect v7.0.1 container images (see below). My attempt to overwrite the following Kafka Connect configuration defaults using env variables was unsuccessful:

• max.request.size (producer)
• max.partition.fetch.bytes (consumer)
• auto.offset.reset (consumer)

I exec'd into the Kafka Connect container and was able to verify the above configs appeared in the connect properties file (with the producer./consumer. prefix and overriden values):

[appuser@0d12890fb402 ~]$ cat /etc/kafka-connect/kafka-connect.properties
group.id=kafka-connect
producer.max.request.size=10485880 <------------------------------------------------------ HERE
status.storage.replication.factor=1
key.converter=org.apache.kafka.connect.storage.StringConverter
config.storage.topic=kafka-connect-configs
log4j.loggers=org.apache.kafka.connect=DEBUG
offset.storage.replication.factor=1
plugin.path=/usr/share/java,/usr/share/confluent-hub-components/,/connectors/
offset.storage.topic=kafka-connect-offsets
rest.advertised.port=8083
bootstrap.servers=kafka:29092
value.converter=org.apache.kafka.connect.json.JsonConverter
rest.advertised.host.name=kafka-connect
rest.port=8083
status.storage.topic=kafka-connect-status
consumer.max.partition.fetch.bytes=10485880 <-------------------------------------------- HERE
consumer.auto.offset.reset=latest <------------------------------------------------------ HERE
config.storage.replication.factor=1

However, when I look at the Kafka Connect container logs, I still see the default values:

[2022-05-03 11:52:49,397] INFO ProducerConfig values: 
    ...
    max.request.size = 1048576
    ...

...

[2022-05-03 11:52:49,509] INFO ConsumerConfig values: 
    ...
    auto.offset.reset = earliest
         ...
    max.partition.fetch.bytes = 1048576
    ...

Any help here would be appreciated. I'm currently out of ideas after reading through the docs and past related issues.

References:

• https://github.com/confluentinc/cp-docker-images/issues/445
• https://docs.confluent.io/platform/current/installation/docker/config-reference.html#confluent-ak-configuration
• https://docs.confluent.io/3.3.0/connect/userguide.html#overriding-producer-consumer-settings

docker-compose.yaml

version: '3'

services:

  zookeeper:
    image: confluentinc/cp-zookeeper:7.0.1
    container_name: zookeeper
    ports:
      - 2181:2181
    environment:
      ZOOKEEPER_CLIENT_PORT : 2181
      ZOOKEEPER_TICK_TIME   : 2000

  kafka:
    image: confluentinc/cp-kafka:7.0.1
    container_name: kafka
    depends_on:
      - zookeeper
    hostname: kafka
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID                                : 1
      KAFKA_ZOOKEEPER_CONNECT                        : zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP           : PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS                     : PLAINTEXT://${KAFKAHOST}:9092,PLAINTEXT_INTERNAL://kafka:29092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR         : 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR            : 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR : 1
      KAFKA_MESSAGE_MAX_BYTES                        : 10485880

  schema-registry:
    image: confluentinc/cp-schema-registry:7.0.1
    container_name: schema-registry
    depends_on:
      - kafka
    hostname: schema-registry
    ports:
      - 8085:8085
    environment:
      SCHEMA_REGISTRY_HOST_NAME                     : schema-registry
      SCHEMA_REGISTRY_LISTENERS                     : http://0.0.0.0:8085
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS  : kafka:29092

  kafka-connect:
    image: confluentinc/cp-kafka-connect:7.0.1
    container_name: kafka-connect
    ports:
      - 8083:8083
    depends_on:
      - kafka
      - schema-registry
      - zookeeper
    environment:
      CONNECT_BOOTSTRAP_SERVERS                   : kafka:29092
      CONNECT_REST_ADVERTISED_HOST_NAME           : kafka-connect
      CONNECT_REST_ADVERTISED_PORT                : 8083
      CONNECT_REST_PORT                           : 8083
      CONNECT_GROUP_ID                            : kafka-connect

      CONNECT_PLUGIN_PATH                         : /usr/share/java,/usr/share/confluent-hub-components/,/connectors/
      CONNECT_LOG4J_LOGGERS                       : org.apache.kafka.connect=DEBUG

      CONNECT_CONFIG_STORAGE_TOPIC                : kafka-connect-configs
      CONNECT_OFFSET_STORAGE_TOPIC                : kafka-connect-offsets
      CONNECT_STATUS_STORAGE_TOPIC                : kafka-connect-status

      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR   : 1
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR   : 1
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR   : 1

      CONNECT_KEY_CONVERTER                       : org.apache.kafka.connect.storage.StringConverter
      CONNECT_VALUE_CONVERTER                     : org.apache.kafka.connect.json.JsonConverter

      # Override producer config
      CONNECT_PRODUCER_MAX_REQUEST_SIZE           : 10485880 # <-------------------------------------------- HERE

      # Override consumer configs
      CONNECT_CONSUMER_MAX_PARTITION_FETCH_BYTES  : 10485880 # <-------------------------------------------- HERE
      CONNECT_CONSUMER_AUTO_OFFSET_RESET          : latest   # <-------------------------------------------- HERE

    command:
      - bash
      - -c
      - |
        #
        echo "Launching Kafka Connect worker"
        confluent-hub install --no-prompt confluentinc/kafka-connect-s3:10.0.5
        confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:10.3.3
        /etc/confluent/docker/run &
        #
        sleep infinity

[spruisken]

So I managed to override individual connector configs by setting connector.client.config.override.policy=All and adding consumer.override.auto.offset.reset=latest, ... in the connector config. This worked and solved my problem. See https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration.

[artinzamani]

You sir, are a savior. I don't know how to thank you.