Can't override successfully in Kafka Connect 2.4 my connectors

Hi i'm looking to use the new Override policy, that have been released in 2.3, through java code.

And i want to create an example like this :

• Create a Topic with 10 messages

• Create a Consumer that consume Messages and next send them to a default FileSink

• Create an Override Sink that should not take the data from the Consumer (It's configured as Earliest)

• Produce a message that is Consume and take by the two Sinks !

Here are the configurations of my SINK (file) connectors (the default one) :

        taskOut = new FileStreamSinkTask();
        Map<String, String> sinkProperties = new HashMap<>();
        sinkProperties.put(FileStreamSinkConnector.TOPICS_CONFIG, new ConstantSettingsBehavior().SINGLE_TOPIC);
        sinkProperties.put(FileStreamSinkConnector.FILE_CONFIG, new ConstantSettingsBehavior().FILE_OUT_LATEST);
        sinkProperties.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(5_000));
        sinkProperties.put(ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        connectorOut.start(sinkProperties);
        taskOut.initialize(createMock(SinkTaskContext.class));
        taskOut.start(connectorOut.taskConfigs(1).get(0));

And here the Earliest (only what is changing) :

     sinkProperties.put(FileStreamSinkConnector.TOPICS_CONFIG, new ConstantSettingsBehavior().SINGLE_TOPIC);
        sinkProperties.put(FileStreamSinkConnector.FILE_CONFIG, new ConstantSettingsBehavior().FILE_OUT_EARLY);
        sinkProperties.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(5_000));
        sinkProperties.put(ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

Next i'm going to create a consumer that take the messages from the Topic as
List< SinkRecord >

I give this list to the tasks of each connectors :

        myLatestOne.getTaskOut().put(data);
        myEarlyOne.getTaskOut().put(data);

But it looks that i'm not doing the right way ! Because all messages are taken by each connectors

Here the code the pull Request of the code Override code that i'm using.

If i miss something don't hesitate to tell me ! (first question).

Thank's

So i gave to do it trough out JAVA. I found a way to do with terminal that is pretty easy :

Command to do

We first launch our server Zookeeper :

 bin/zookeeper-server-start.sh config/zokeeper.properties

Next we start our server kafka :

bin/kafka-server-start.sh config/server.properties

We need to create a topic :

./bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

Now we need to produce messages :

./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
> [Your message]

And now we can launch our worker, with 1 connector connected.
You can have their properties in the config file.

bootstrap.servers=localhost:9092

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

key.converter.schemas.enable=true
value.converter.schemas.enable=true

offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000
connector.client.config.override.policy=All 

connector.client.config.override.policy=All Allow to override the client by the connector.

Here is our connector with the option earliest (If there is no offset saved it start from the first entry)

name=local-file-earliest-sink 
connector.class=org.apache.kafka.connect.file.FileStreamSinkConnector
tasks.max=1
file=/tmp/test.sink.earliest.txt
topics=test
consumer.override.auto.offset.reset=earliest
value.converter=org.apache.kafka.connect.storage.StringConverter

sudo ./bin/connect-standalone.sh config/connect-standalone.properties  config/connect-file-sink-early.properties

We stop it few seconds later (you can look at tmp/test.sink.earliest.txt ).

This time we add a new connector :

name=local-file-latest-sink
connector.class=org.apache.kafka.connect.file.FileStreamSinkConnector
tasks.max=1
file=/tmp/test.sink.latest.txt
topics=test
consumer.override.auto.offset.reset=latest
value.converter=org.apache.kafka.connect.storage.StringConverter

We can launch both of them :

sudo ./bin/connect-standalone.sh config/connect-standalone.properties  config/connect-file-sink-early.properties config/connect-file-sink-latest.properties 

We can add new messages and check if /tmp/test.sink.latest.txt is only fill with those messages.

Explanation

The main idea here is to be able to have a default reconfigurable each connector in a different way. So to do it we are using the add Override Policy

Each connector will create a new consumer group ID. If they both read from the same topics, then they will both get all messages

Also, consumer and producer overrides have already been possible at the worker level, and I've not seen anyone write their own connector like this since you could just use connect-standalone