Common Worker Configuration Options

You can set common worker configuration options for standalone or distributed mode in the connect-<standalone/distributed>.properties file. The options control basic functionality, including which cluster to communicate with and data format.

Setting the Schema Registry URL for the Avro Converter

Set the Schema Registry URL for the converter through the following properties:
  • key.converter.schema.registry.url=<URL:PORT>
  • value.converter.schema.registry.url=<URL:PORT>
  • For Avro, use io.confluent.connect.avro.AvroConverter.
  • For Protobuf, use io.confluent.connect.protobuf.ProtobufConverter).
  • For JSON Schema, use io.confluent.connect.json.JsonSchemaConverter.

If you do not set these properties, the Schema Registry URL is taken from ZooKeeper.

The following table describes the common worker configuration parameters:

Parameter Description
plugin.path

The comma-separated list of paths to directories that contain Kafka Connect plugins.

  • Type: string
  • Default: empty

key.converter

Converter class for key Connect data. This controls the format of the data that will be written to HPE Ezmeral Data Fabric Streams for source connectors or read from Data Fabric Streams for sink connectors.

  • Type: class
  • Default: empty

value.converter

Converter class for value Connect data. This controls the format of the data that will be written to HPE Ezmeral Data Fabric Streams for source connectors or read from Data Fabric Streams for sink connectors.

  • Type: class
  • Default: empty

internal.key.converter

Converter class for internal key Connect data that implements the Converter interface. Used for converting data like offsets and configs.

  • Type: class
  • Default:

internal.value.converter

Converter class for offset value Connect data that implements the Converter interface. Used for converting data like offsets and configs.

  • Type: class
  • Default:

offset.flush.interval.ms

Interval (milliseconds) at which to try committing offsets for tasks.

  • Type: long
  • Default: 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

rest.advertised.host.name

If set, this is the hostname that will be given out to other workers to connect to.

  • Type: string

rest.advertised.port

If set, this is the port that will be given out to other workers to connect to.

  • Type: int

rest.host.name

Hostname for the REST API. If this is set, it will only bind to this interface.

  • Type: string

rest.port

Port for the REST API to listen on.

  • Type: int
  • Default: 8083

task.shutdown.graceful.timeout.ms

Amount of time to wait (milliseconds) 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
streams.consumer.streams.default.stream If set, topic names can be used in the Sink task configuration without the stream name. The defined default stream is used.
streams.producer.producer.default.stream If set, topic names can be used in the Source task configuration without the stream name. The defined default stream is used.