docker-compose.yml

#------------------------------------------------------------------------------
# Copyright 2019 Robert Cowart
# 
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# 
#     http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#------------------------------------------------------------------------------

version: '3'
services:
  cp-kafka-connect:
    image: robcowart/cp-kafka-connect-custom:6.0.0_1.0.3_community
    container_name: cp-kafka-connect
    restart: unless-stopped
    network_mode: bridge
    ports:
      - 8083:8083/tcp
    environment:
      ########## General ##########

      # group.id
      #   A unique string that identifies the Connect cluster group this worker belongs to.
      # Type: string
      CONNECT_GROUP_ID: 'connect-cluster'

      # client.id
      #   An id string to pass to the server when making requests. The purpose of this is to be able to track the
      #   source of requests beyond just ip/port by allowing a logical application name to be included in server-side
      #   request logging.
      # Type: string
      # Default: ""
      #CONNECT_CLIENT_ID: ''

      # bootstrap.servers
      #   A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. The client
      #   will make use of all servers irrespective of which servers are specified here for bootstrapping—this list
      #   only impacts the initial hosts used to discover the full set of servers. This list should be in the form
      #   host1:port1,host2:port2,.... Since these servers are just used for the initial connection to discover the
      #   full cluster membership (which may change dynamically), this list need not contain the full set of servers
      #   (you may want more than one, though, in case a server is down).
      # Type: list
      # Default: localhost:9092
      CONNECT_BOOTSTRAP_SERVERS: '192.2.0.11:9092,192.2.0.12:9092,192.2.0.13:9092'
      
      # security.protocol
      #   Protocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.
      # Type: string
      # Default: PLAINTEXT
      CONNECT_SECURITY_PROTOCOL: 'PLAINTEXT'
      
      # plugin.path
      #   List of paths separated by commas (,) that contain plugins (connectors, converters, transformations). The
      #   list should consist of top level directories that include any combination of: a) directories immediately
      #   containing jars with plugins and their dependencies b) uber-jars with plugins and their dependencies c)
      #   directories immediately containing the package directory structure of classes of plugins and their
      #   dependencies Note: symlinks will be followed to discover dependencies or plugins. Examples:
      #   plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors
      # Type: list
      # Default: null
      CONNECT_PLUGIN_PATH: '/usr/share/java,/usr/share/confluent-hub-components'
      
      # connect.protocol
      #   Compatibility mode for Kafka Connect Protocol
      # Type: string
      # Default: compatible [eager, compatible]
      #CONNECT_CONNECT_PROTOCOL: 'compatible'
      
      
      
      ########## REST ##########

      # listeners
      #   List of comma-separated URIs the REST API will listen on. The supported protocols are HTTP and HTTPS. Specify
      #   hostname as 0.0.0.0 to bind to all interfaces. Leave hostname empty to bind to default interface. Examples of
      #   legal listener lists: HTTP://myhost:8083,HTTPS://myhost:8084
      # Type: list
      # Default: null
      CONNECT_LISTENERS: 'http://0.0.0.0:8083'
      
      # rest.host.name
      #   Hostname for the REST API. If this is set, it will only bind to this interface.
      # Type: string
      # Default: null
      CONNECT_REST_HOST_NAME: '0.0.0.0'
      
      # rest.port
      #   Port for the REST API to listen on.
      # Type: int
      # Default: 8083
      CONNECT_REST_PORT: 8083
      

      # rest.advertised.listener
      #   Sets the advertised listener (HTTP or HTTPS) which will be given to other workers to use.
      # Type: string
      # Default: null
      CONNECT_REST_ADVERTISED_LISTENER: 'http'

      # rest.advertised.host.name
      #   If this is set, this is the hostname that will be given out to other workers to connect to.
      # Type: string
      # Default: null
      CONNECT_REST_ADVERTISED_HOST_NAME: '192.2.0.11'
      
      # rest.advertised.port
      #   If this is set, this is the port that will be given out to other workers to connect to.
      # Type: int
      # Default: null
      CONNECT_REST_ADVERTISED_PORT: 8083
      

      # rest.extension.classes
      #   Comma-separated names of ConnectRestExtension classes, loaded and called in the order specified. Implementing
      #   the interface ConnectRestExtension allows you to inject into Connect's REST API user defined resources like
      #   filters. Typically used to add custom capability like logging, security, etc.
      # Type: list
      # Default: ""
      #CONNECT_REST_EXTENSION_CLASSES: ''
      
      
      
      ########## Buffers ##########

      # receive.buffer.bytes
      #   The size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default
      #   will be used.
      # Type: int
      # Default: 32768 [0,...]
      #CONNECT_RECEIVER_BUFFER_BYTES: 32768
      
      # send.buffer.bytes
      #   The size of the TCP send buffer (SO_SNDBUF) to use when sending data. If the value is -1, the OS default will
      #   be used.
      # Type: int
      # Default: 131072 [0,...]
      #CONNECT_SEND_BUFFER_BYTES: 131072
      
      
      
      ########## Storage Topics ##########

      # config.storage.topic
      #   The name of the Kafka topic where connector configurations are stored
      # Type: string
      CONNECT_CONFIG_STORAGE_TOPIC: 'connect-config'
      
      # config.storage.replication.factor
      #   Replication factor used when creating the configuration storage topic
      # Type: short
      # Default: 3 [1,...]
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 2
      
      
      # offset.storage.topic
      #   The name of the Kafka topic where connector offsets are stored
      # Type: string
      CONNECT_OFFSET_STORAGE_TOPIC: 'connect-offset'
      
      # offset.storage.partitions
      #   The number of partitions used when creating the offset storage topic
      # Type: int
      # Default: 25 [1,...]
      CONNECT_OFFSET_STORAGE_PARTITIONS: 12
      
      # offset.storage.replication.factor
      #   Replication factor used when creating the offset storage topic
      # Type: short
      # Default: 3 [1,...]
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 2
      
      # offset.flush.interval.ms
      #   Interval at which to try committing offsets for tasks.
      # Type: long
      # Default: 60000
      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 60000
      
      # offset.flush.timeout.ms
      #   Maximum number of milliseconds to wait for records to flush and partition offset data to be committed to
      #   offset storage before cancelling the process and restoring the offset data to be committed in a future
      #   attempt.
      # Type: long
      # Default: 5000
      CONNECT_OFFSET_FLUSH_TIMEOUT_MS: 5000

      # offset.storage.file.filename
      #   The file to store connector offsets in. By storing offsets on disk, a standalone process can be stopped and
      #   started on a single node and resume where it previously left off.
      # Type: string
      # Default: ""
      #CONNECT_OFFSET_STORAGE_FILE_FILENAME: ''
      
      # status.storage.topic
      #   The name of the Kafka topic where connector and task status are stored
      # Type: string
      CONNECT_STATUS_STORAGE_TOPIC: 'connect-status'
      
      # status.storage.partitions
      #   The number of partitions used when creating the status storage topic
      # Type: int
      # Default: 5 [1,...]
      CONNECT_STATUS_STORAGE_PARTITIONS: 3
      
      # status.storage.replication.factor
      #   Replication factor used when creating the status storage topic
      # Type: short
      # Default: 3 [1,...]
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 2
      
      

      ########## Converters ##########

      # key.converter
      #   Converter class used to convert between Kafka Connect format and the serialized form that is written to
      #   Kafka. This controls the format of the keys in messages written to or read from Kafka, and since this is
      #   independent of connectors it allows any connector to work with any serialization format. Examples of common
      #   formats include JSON and Avro.
      # Type: class
      CONNECT_KEY_CONVERTER: 'org.apache.kafka.connect.json.JsonConverter'
      
      # value.converter
      #   Converter class used to convert between Kafka Connect format and the serialized form that is written to
      #   Kafka. This controls the format of the values in messages written to or read from Kafka, and since this is
      #   independent of connectors it allows any connector to work with any serialization format. Examples of common
      #   formats include JSON and Avro.
      # Type: class
      CONNECT_VALUE_CONVERTER: 'org.apache.kafka.connect.json.JsonConverter'
      
      # internal.key.converter
      #   Converter class used to convert between Kafka Connect format and the serialized form that is written to
      #   Kafka. This controls the format of the keys in messages written to or read from Kafka, and since this is
      #   independent of connectors it allows any connector to work with any serialization format. Examples of common
      #   formats include JSON and Avro. This setting controls the format used for internal bookkeeping data used by
      #   the framework, such as configs and offsets, so users can typically use any functioning Converter
      #   implementation. Deprecated; will be removed in an upcoming version.
      # Type: class
      # Default: org.apache.kafka.connect.json.JsonConverter
      CONNECT_INTERNAL_KEY_CONVERTER: 'org.apache.kafka.connect.json.JsonConverter'
      
      # internal.value.converter
      #   Converter class used to convert between Kafka Connect format and the serialized form that is written to
      #   Kafka. This controls the format of the values in messages written to or read from Kafka, and since this is
      #   independent of connectors it allows any connector to work with any serialization format. Examples of common
      #   formats include JSON and Avro. This setting controls the format used for internal bookkeeping data used by
      #   the framework, such as configs and offsets, so users can typically use any functioning Converter
      #   implementation. Deprecated; will be removed in an upcoming version.
      # Type: class
      # Default: org.apache.kafka.connect.json.JsonConverter
      CONNECT_INTERNAL_VALUE_CONVERTER: 'org.apache.kafka.connect.json.JsonConverter'
      
      
      
      ########## Access Control ##########

      # access.control.allow.methods
      #   Sets the methods supported for cross origin requests by setting the Access-Control-Allow-Methods header. The
      #   default value of the Access-Control-Allow-Methods header allows cross origin requests for GET, POST and HEAD.
      # Type: string
      # Default: ""
      #CONNECT_CONTROL_ALLOW_METHODS: ''
      
      # access.control.allow.origin
      #   Value to set the Access-Control-Allow-Origin header to for REST API requests.To enable cross origin access,
      #   set this to the domain of the application that should be permitted to access the API, or '*' to allow access
      #   from any domain. The default value only allows access from the domain of the REST API.
      # Type: string
      # Default: ""
      #CONNECT_CONTROL_ALLOW_ORIGINS: ''


      
      ########## SSL ##########

      # ssl.protocol
      #   The SSL protocol used to generate the SSLContext. Default setting is TLS, which is fine for most cases.
      #   Allowed values in recent JVMs are TLS, TLSv1.1 and TLSv1.2. SSL, SSLv2 and SSLv3 may be supported in older
      #   JVMs, but their usage is discouraged due to known security vulnerabilities.
      # Type: string
      # Default: TLS
      #CONNECT_SSL_PROTOCOL: 'TLS'
      
      # ssl.enabled.protocols
      #   The list of protocols enabled for SSL connections.
      # Type: list
      # Default: TLSv1.2,TLSv1.1,TLSv1
      #CONNECT_SSL_ENABLED_PROTOCOLS: 'TLSv1.2,TLSv1.1,TLSv1'
      
      # ssl.provider
      #   The name of the security provider used for SSL connections. Default value is the default security provider of
      #   the JVM.
      # Type: string
      # Default: null
      #CONNECT_SSL_PROVIDER: ''
      
      # ssl.cipher.suites
      #   A list of cipher suites. This is a named combination of authentication, encryption, MAC and key exchange
      #   algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol.
      #   By default all the available cipher suites are supported.
      # Type: list
      # Default: null
      #CONNECT_SSL_CIPHER_SUITES: ''
      
      # ssl.key.password
      #   The password of the private key in the key store file. This is optional for client.
      # Type: password
      # Default: null
      #CONNECT_SSL_KEY_PASSOWRD: ''
      
      # ssl.keystore.location
      #   The location of the key store file. This is optional for client and can be used for two-way authentication
      #   for client.
      # Type: string
      # Default: null
      #CONNECT_SSL_KEYSTORE_LOCATION: ''
      
      # ssl.keystore.password
      #   The store password for the key store file. This is optional for client and only needed if
      #   ssl.keystore.location is configured.
      # Type: password
      # Default: null
      #CONNECT_SSL_KEYSTORE_PASSWORD: ''
      
      # ssl.keystore.type
      #   The file format of the key store file. This is optional for client.
      # Type: string
      # Default: JKS
      #CONNECT_SSL_KEYSTORE_TYPE: 'JKS'
      
      # ssl.truststore.location
      #   The location of the trust store file.
      # Type: string
      # Default: null
      #CONNECT_SSL_TRUSTSTORE_LOCATION: ''
      
      # ssl.truststore.password
      #   The password for the trust store file. If a password is not set access to the truststore is still available,
      #   but integrity checking is disabled.
      # Type: password
      # Default: null
      #CONNECT_SSL_TRUSTSTORE_PASSWORD: ''
      
      # ssl.truststore.type
      #   The file format of the trust store file.
      # Type: string
      # Default: JKS
      #CONNECT_SSL_TRUSTSTORE_TYPE: 'JKS'
      
      # ssl.keymanager.algorithm
      #   The algorithm used by key manager factory for SSL connections. Default value is the key manager factory
      #   algorithm configured for the Java Virtual Machine.
      # Type: string
      # Default: SunX509
      #CONNECT_SSL_KEYMANAGER_ALGORITHM: 'SunX509'
      
      # ssl.trustmanager.algorithm
      #   The algorithm used by trust manager factory for SSL connections. Default value is the trust manager factory
      #   algorithm configured for the Java Virtual Machine.
      # Type: string
      # Default: PKIX
      #CONNECT_SSL_TRUSTMANAGER_ALGORITHM: 'PKIX'
      
      # ssl.client.auth
      #   Configures kafka broker to request client authentication. The following settings are common:
      #     required - If set to required client authentication is required.
      #     requested - This means client authentication is optional. unlike requested, if this option is set client
      #                 can choose not to provide authentication information about itself
      #     none - This means client authentication is not needed.
      # Type: string
      # Default: none
      #CONNECT_SSL_CLIENT_AUTH: 'none'
      
      # ssl.endpoint.identification.algorithm
      #   The endpoint identification algorithm to validate server hostname using server certificate.
      # Type: string
      # Default: https
      #CONNECT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM: 'https'
      
      # ssl.secure.random.implementation
      #   The SecureRandom PRNG implementation to use for SSL cryptography operations.
      # Type: string
      # Default: null
      #CONNECT_SSL_SECURE_RANDOM_IMPLEMENTATION: ''
      
      
      
      ########## SASL ##########

      # sasl.client.callback.handler.class
      #   The fully qualified name of a SASL client callback handler class that implements the
      #   AuthenticateCallbackHandler interface.
      # Type: class
      # Default: null
      #CONNECT_SASL_CLIENT_CALLBACK_HANDLER_CLASS: ''
      
      # sasl.jaas.config
      #   JAAS login context parameters for SASL connections in the format used by JAAS configuration files. JAAS
      #   configuration file format is described here. The format for the value is: 'loginModuleClass controlFlag
      #   (optionName=optionValue)*;'. For brokers, the config must be prefixed with listener prefix and SASL mechanism
      #   name in lower-case. For example,
      #   listener.name.sasl_ssl.scram-sha-256.sasl.jaas.config=com.example.ScramLoginModule required;
      # Type: password
      # Default: null
      #CONNECT_SASL_JAAS_CONFIG: ''
      
      # sasl.mechanism
      #   SASL mechanism used for client connections. This may be any mechanism for which a security provider is
      #   available. GSSAPI is the default mechanism.
      # Type: string
      # Default: GSSAPI
      #CONNECT_SASL_MECHANISM: 'GSSAPI'
      
      # sasl.kerberos.service.name
      #   The Kerberos principal name that Kafka runs as. This can be defined either in Kafka's JAAS config or in
      #   Kafka's config.
      # Type: string
      # Default: null
      #CONNECT_SASL_KERBEROS_SERVICE_NAME: ''
      
      # sasl.kerberos.kinit.cmd
      #   Kerberos kinit command path.
      # Type: string
      # Default: /usr/bin/kinit
      #CONNECT_SASL_KERBEROS_KINIT_CMD: '/usr/bin/kinit'
      
      # sasl.kerberos.min.time.before.relogin
      #   Login thread sleep time between refresh attempts.
      # Type: long
      # Default: 60000
      #CONNECT_SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN: 60000
      
      # sasl.kerberos.ticket.renew.jitter
      #   Percentage of random jitter added to the renewal time.
      # Type: double
      # Default: 0.05
      #CONNECT_SASL_KERBEROS_TICKET_RENEW_JITTER: 0.05
      
      # sasl.kerberos.ticket.renew.window.factor
      #   Login thread will sleep until the specified window factor of time from last refresh to ticket's expiry has
      #   been reached, at which time it will try to renew the ticket.
      # Type: double
      # Default: 0.8
      #CONNECT_SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR: 0.8

      # sasl.login.callback.handler.class
      #   The fully qualified name of a SASL login callback handler class that implements the
      #   AuthenticateCallbackHandler interface. For brokers, login callback handler config must be prefixed with
      #   listener prefix and SASL mechanism name in lower-case. For example,
      #   listener.name.sasl_ssl.scram-sha-256.sasl.login.callback.handler.class=com.example.CustomScramLoginCallbackHandler
      # Type: class
      # Default: null
      #CONNECT_SASL_LOGIN_CALLBACK_HANDLER_CLASS: ''
      
      # sasl.login.class
      #   The fully qualified name of a class that implements the Login interface. For brokers, login config must be
      #   prefixed with listener prefix and SASL mechanism name in lower-case. For example,
      #   listener.name.sasl_ssl.scram-sha-256.sasl.login.class=com.example.CustomScramLogin
      # Type: class
      # Default: null
      #CONNECT_SASL_LOGIN_CLASS: ''
      
      # sasl.login.refresh.buffer.seconds
      #   The amount of buffer time before credential expiration to maintain when refreshing a credential, in seconds.
      #   If a refresh would otherwise occur closer to expiration than the number of buffer seconds then the refresh
      #   will be moved up to maintain as much of the buffer time as possible. Legal values are between 0 and 3600
      #   (1 hour); a default value of 300 (5 minutes) is used if no value is specified. This value and
      #   sasl.login.refresh.min.period.seconds are both ignored if their sum exceeds the remaining lifetime of a
      #   credential. Currently applies only to OAUTHBEARER.
      # Type: short
      # Default: 300 [0,...,3600]
      #CONNECT_SASL_LOGIN_REFRESH_BUFFER_SECONDS: 300
      
      # sasl.login.refresh.min.period.seconds
      #   The desired minimum time for the login refresh thread to wait before refreshing a credential, in seconds.
      #   Legal values are between 0 and 900 (15 minutes); a default value of 60 (1 minute) is used if no value is
      #   specified. This value and sasl.login.refresh.buffer.seconds are both ignored if their sum exceeds the
      #   remaining lifetime of a credential. Currently applies only to OAUTHBEARER.
      # Type: short
      # Default: 60 [0,...,900]
      #CONNECT_SASL_LOGIN_REFRESH_MIN_PERIOD_SECONDS: 60
      
      # sasl.login.refresh.window.factor
      #   Login refresh thread will sleep until the specified window factor relative to the credential's lifetime has
      #   been reached, at which time it will try to refresh the credential. Legal values are between 0.5 (50%) and 1.0
      #   (100%) inclusive; a default value of 0.8 (80%) is used if no value is specified. Currently applies only to
      #   OAUTHBEARER.
      # Type: double
      # Default: 0.8 [0.5,...,1.0]
      #CONNECT_SASL_LOGIN_REFRESH_WINDOW_FACTOR: 0.8
      
      # sasl.login.refresh.window.jitter
      #   The maximum amount of random jitter relative to the credential's lifetime that is added to the login refresh
      #   thread's sleep time. Legal values are between 0 and 0.25 (25%) inclusive; a default value of 0.05 (5%) is
      #   used if no value is specified. Currently applies only to OAUTHBEARER.
      # Type: double
      # Default: 0.05 [0.0,...,0.25]
      #CONNECT_SASL_LOGIN_REFRESH_WINDOW_JITTER: 0.05
      


      ########## Timeouts and Backoffs ##########
      
      # connections.max.idle.ms
      #   Close idle connections after the number of milliseconds specified by this config.
      # Type: long
      # Default: 540000
      #CONNECT_CONNECTIONS_MAX_IDLE_MS: 540000
      
      # heartbeat.interval.ms
      #   The expected time between heartbeats to the group coordinator when using Kafka's group management facilities.
      #   Heartbeats are used to ensure that the worker's session stays active and to facilitate rebalancing when new
      #   members join or leave the group. The value must be set lower than session.timeout.ms, but typically should be
      #   set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal
      #   rebalances.
      # Type: int
      # Default: 3000
      #CONNECT_HEARTBEAT_INTERVAL_MS: 3000
      
      # metadata.max.age.ms
      #   The period of time in milliseconds after which we force a refresh of metadata even if we haven't seen any
      #   partition leadership changes to proactively discover any new brokers or partitions.
      # Type: long
      # Default: 300000 [0,...]
      #CONNECT_METADATA_MAX_AGE_MS: 300000
      
      # rebalance.timeout.ms
      #   The maximum allowed time for each worker to join the group once a rebalance has begun. This is basically a
      #   limit on the amount of time needed for all tasks to flush any pending data and commit offsets. If the timeout
      #   is exceeded, then the worker will be removed from the group, which will cause offset commit failures.
      # Type: int
      # Default: 60000
      #CONNECT_REBALANCE_TIMEOUT_MS: 60000
      
      # reconnect.backoff.max.ms
      #   The maximum amount of time in milliseconds to wait when reconnecting to a broker that has repeatedly failed
      #   to connect. If provided, the backoff per host will increase exponentially for each consecutive connection
      #   failure, up to this maximum. After calculating the backoff increase, 20% random jitter is added to avoid
      #   connection storms.
      # Type: long
      # Default: 1000 [0,...]
      #CONNECT_RECONNECT_BACKOFF_MAX_MS: 1000
      
      # reconnect.backoff.ms
      #   The base amount of time to wait before attempting to reconnect to a given host. This avoids repeatedly
      #   connecting to a host in a tight loop. This backoff applies to all connection attempts by the client to a
      #   broker.
      # Type: long
      # Default: 50 [0,...]
      #CONNECT_RECONNECT_BACKOFF_MS: 50
      
      # request.timeout.ms
      #   The configuration controls the maximum amount of time the client will wait for the response of a request. If
      #   the response is not received before the timeout elapses the client will resend the request if necessary or
      #   fail the request if retries are exhausted.
      # Type: int
      # Default: 40000 [0,...]
      #CONNECT_REQUEST_TIMEOUT_MS: 40000
      
      # retry.backoff.ms
      #   The amount of time to wait before attempting to retry a failed request to a given topic partition. This
      #   avoids repeatedly sending requests in a tight loop under some failure scenarios.
      # Type: long
      # Default: 100 [0,...]
      #CONNECT_RETRY_BACKOFF_MS: 100
      
      # scheduled.rebalance.max.delay.ms
      #   Compatibility mode for Kafka Connect Protocol
      # Type: int
      # Default: 300000 [0,...,2147483647]
      #CONNECT_SCHEDULED_REBALANCE_MAX_DELAY_MS: 300000
      
      # session.timeout.ms
      #   The timeout used to detect worker failures. The worker sends periodic heartbeats to indicate its liveness to
      #   the broker. If no heartbeats are received by the broker before the expiration of this session timeout, then
      #   the broker will remove the worker from the group and initiate a rebalance. Note that the value must be in the
      #   allowable range as configured in the broker configuration by group.min.session.timeout.ms and
      #   group.max.session.timeout.ms.
      # Type: int
      # Default: 10000
      #CONNECT_SESSION_TIMEOUT_MS: 10000
      
      # task.shutdown.graceful.timeout.ms
      #   Amount of time to wait for tasks to shutdown gracefully. This is the total amount of time, not per task. All
      #   task have shutdown triggered, then they are waited on sequentially.
      # Type: long
      # Default: 5000
      #CONNECT_TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS: 5000
      
      # worker.sync.timeout.ms
      #   When the worker is out of sync with other workers and needs to resynchronize configurations, wait up to this
      #   amount of time before giving up, leaving the group, and waiting a backoff period before rejoining.
      # Type: int
      # Default: 3000
      #CONNECT_WORKER_SYNC_TIMEOUT_MS: 3000
      
      # worker.unsync.backoff.ms
      #   When the worker is out of sync with other workers and fails to catch up within worker.sync.timeout.ms, leave
      #   the Connect cluster for this long before rejoining.
      # Type: int
      # Default: 300000
      #CONNECT_WORKER_UNSYNC_BACKOFF_MS: 300000
      


      ########## Additional Configuration ##########
      
      # client.dns.lookup
      #   Controls how the client uses DNS lookups. If set to use_all_dns_ips then, when the lookup returns multiple IP
      #   addresses for a hostname, they will all be attempted to connect to before failing the connection. Applies to
      #   both bootstrap and advertised servers. If the value is resolve_canonical_bootstrap_servers_only each entry
      #   will be resolved and expanded into a list of canonical names.
      # Type: string
      # Default: default [default, use_all_dns_ips, resolve_canonical_bootstrap_servers_only]
      #CONNECT_DNS_LOOKUP: 'default'
      
      # config.providers
      #   Comma-separated names of ConfigProvider classes, loaded and used in the order specified. Implementing the
      #   interface ConfigProvider allows you to replace variable references in connector configurations, such as for
      #   externalized secrets.
      # Type: list
      # Default: ""
      #CONNECT_CONFIG_PROVIDERS: ''
      
      # connector.client.config.override.policy
      #   Class name or alias of implementation of ConnectorClientConfigOverridePolicy. Defines what client
      #   configurations can be overriden by the connector. The default implementation is `None`. The other possible
      #   policies in the framework include `All` and `Principal`.
      # Type: string
      # Default: None
      #CONNECT_CONNECTOR_CLIENT_CONFIG_OVERRIDE_POLICY: 'None'
      
      # header.converter
      #   HeaderConverter class used to convert between Kafka Connect format and the serialized form that is written to
      #   Kafka. This controls the format of the header values in messages written to or read from Kafka, and since
      #   this is independent of connectors it allows any connector to work with any serialization format. Examples of
      #   common formats include JSON and Avro. By default, the SimpleHeaderConverter is used to serialize header
      #   values to strings and deserialize them by inferring the schemas.
      # Type: class
      # Default: org.apache.kafka.connect.storage.SimpleHeaderConverter
      #CONNECT_HEADER_CONVERTER: 'org.apache.kafka.connect.storage.SimpleHeaderConverter'
      
      
      
      ########## Metric Reporter ##########

      # metric.reporters
      #   A list of classes to use as metrics reporters. Implementing the
      #   org.apache.kafka.common.metrics.MetricsReporter interface allows plugging in classes that will be notified of
      #   new metric creation. The JmxReporter is always included to register JMX statistics.
      # Type: list
      # Default: ""
      #CONNECT_METRIC_REPORTERS: ''
      
      # metrics.num.samples
      #   The number of samples maintained to compute metrics.
      # Type: int
      # Default: 2 [1,...]
      #CONNECT_METRICS_NUM_SAMPLES: 2
      
      # metrics.recording.level
      #   The highest recording level for metrics.
      # Type: string
      # Default: INFO [INFO, DEBUG]
      #CONNECT_METRICS_RECORDING_LEVEL: 'INFO'
      
      # metrics.sample.window.ms
      #   The window of time a metrics sample is computed over.
      # Type: long
      # Default: 30000 [0,...]
      #CONNECT_METRICS_SAMPLE_WINDOW_MS: 30000