understanding kafka connect topics and configurations (offset.storage.topic, config.storage.topic, status.storage.topic)

[cstmgl]

Hi all,

I have a problem, which I now have a work around, but I'm trying to figure out the root cause and while doing so understand better if I have some miss configuration on my internal Kafka connect topics.

As part of the resource to define the KafkaConnect we have:
https://strimzi.io/docs/operators/0.26.1/deploying.html#con-kafka-connect-multiple-instances-str

    offset.storage.topic: connect-cluster-offsets (2)
    config.storage.topic: connect-cluster-configs (3)
    status.storage.topic: connect-cluster-status (4)

(2) Kafka topic that stores connector offsets.
(3) Kafka topic that stores connector and task status configurations.
(4) Kafka topic that stores connector and task status updates.

What I experienced is that I updated my connector and had some dependencies to a file in the file system that was always causing the connectors to fail, because they exist in the new configuration but not in the old.

I went and checked the topic itself with the kafka-cli and I noticed I had these configurations:

sh-4.4$ bin/kc.sh -C -t connect-cluster-configs -f 'topic: %t, key: %k, partition: %p, offset: %o, headers: %h\n' -e -q
topic: connect-cluster-configs, key: session-key, partition: 0, offset: 0, headers: 
topic: connect-cluster-configs, key: connect-jdbc-sink, partition: 0, offset: 1, headers: 
topic: connect-cluster-configs, key: task-jdbc-sink-0, partition: 0, offset: 2, headers: 
...
topic: connect-cluster-configs, key: task-jdbc-sink-9, partition: 0, offset: 11, headers: 
topic: connect-cluster-configs, key: commit-jdbc-sink, partition: 0, offset: 12, headers: 
topic: connect-cluster-configs, key: connect-jdbc-sink, partition: 0, offset: 13, headers: 
topic: connect-cluster-configs, key: session-key, partition: 0, offset: 14, headers: 
topic: connect-cluster-configs, key: connect-jdbc-sink, partition: 0, offset: 15, headers: 
topic: connect-cluster-configs, key: connect-jdbc-sink, partition: 0, offset: 16, headers: 
topic: connect-cluster-configs, key: target-state-jdbc-sink, partition: 0, offset: 17, headers:

Now the problem I was experiencing was the fact that it seemed the tasks and the connector itself at each restart would pick up the older configurations and not the ones defined in the "kind: KafkaConnector"
I can see those configurations well defined in the topic itself, but probably the tasks have not moved to the correct consumer group.

My question is, do I need to define any special kind of Kafka policy in these topics so that they always pick up the correct values?
In my case imagine for example I have 2 configurations, one that point to an internal DB and another that point to an external DB and sometimes the tasks don't pick-up my changes to the configurations, I assume because it's picking up the older configurations.

I was wondering if there is some sort of KafkaTopic configuration I need to define so that it prevents this issue.
For example, I notice several entries for the config of the connector itself, with different offsets, wondering if they would expect some sort of Log Compaction based on the Kafka key that would delete old records based on the key updated?

I to be honest could not find if there is a way to configure this at topic level with the topic resource, but I also don't know if this is required/expected

I was looking at the KafkaTopic CRD
https://strimzi.io/docs/operators/0.26.1/deploying.html#con-custom-resources-example-str

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic (1)
metadata:
  name: my-topic
  labels:
    strimzi.io/cluster: my-cluster (2)
spec: (3)
  partitions: 1
  replicas: 1
  config:
    retention.ms: 7200000
    segment.bytes: 1073741824
status:
  conditions: (4)
    lastTransitionTime: "2019-08-20T11:37:00.706Z"
    status: "True"
    type: Ready
  observedGeneration: 1
  / ...

And not only I did not find anything obvious to me that I needed to configure but I would also prefer to not thinker with these defaults.

Can someone maybe give me some hints on how I might be miss configuring these internal topics?

UPDATE:

I have now better explanation on the problem and how did I worked around it, but I'm still confused if this is really the expected results, or I missed something.

In short, the main problem is this:

After I update my cluster and change the configuration of the KafkaConnector it seems the changes do not get propagated to the tasks.
Even after I restart the tasks, they don't pick-up the latest changes.

Say we have a KafkaConnector YAML resource definition, let's call it version 1 that defines X tasks, and they reference a property file located in a file, say

connection.user: ${file:/opt/kafka/external-configuration/connection-details/jdbc-details.properties:db.user}

You update the YAML configuration to version 2 and the application slightly changed (new KafkaConnect version) and now the property will be referenced somewhere else:

    connection.user: ${file:/opt/kafka/external-configuration/connector/jdbc-details.properties:db.user}

After this update the tasks will go into error because it will say that file from version 1 (connection-details/jdbc-details.properties) cannot be found, which it's true, they should now be looking for file of version 2.
So, this leads me to believe that the tasks do not get an updated configuration.

Initially I thought the problem was the connector and the fact that my configurations were not picked up from Kafka.
Now I think the problem is with the actual tasks.

To work around this, I tried to restart the tasks with the Rest API from KafkaConnector but that did not work.

I needed to change the YAML to reduce to say 1 task instead of whatever was defined before.
Let that trigger the tasks re-creation and then increase to my initial number of tasks.
This seems to have worked, but I was wondering why this is needed.

[scholzj]

TBH, I don't really follow the problem from your description. But Kafka Connect manages its topics on its own. So you should not interfere with them normally (+ it also validates some of the topic configurations at startup).

[cstmgl]

I tried deleting both the KafkaConnector and the KafkaConnect and re-creating the resources, but they keep the same.
I would be happy first with a confirmation that it's ok that the Kafka topic for the configs will show all the versions.
Second part is my expectation is that there will be a consumer-group for the connector itself that shows it's consumed up to the last offset of the configs itself. Or would this last assumption not make much sense?

[scholzj]

I would be happy first with a confirmation that it's ok that the Kafka topic for the configs will show all the versions.
Second part is my expectation is that there will be a consumer-group for the connector itself that shows it's consumed up to the last offset of the configs itself. Or would this last assumption not make much sense?

I'm not sure there is any consumer group. I don't think that is not how Connect works. But TBH, this is all connect's internal implementation.

I tried deleting both the KafkaConnector and the KafkaConnect and re-creating the resources, but they keep the same.

Well, I guess you would need to delete the topics (or use different topic names for the new deployment to start with a completely clean slate)

[cstmgl]

I have now better explanation on the problem and how did I worked around it, but I'm still confused if this is really the expected results, or I missed something.

In short the main problem is this:

After I update my cluster and change the configuration of the kafkaconnector it seems the changes do not get propagated to the tasks.
Even after I restart the tasks, they don't pick-up the latest changes.

Say we have a KafkaConnector YAML resource definition, let's call it version 1 that defines X tasks and they reference a property file located in a file, say

connection.user: ${file:/opt/kafka/external-configuration/connection-details/jdbc-details.properties:db.user}

You update the YAML configuration to version 2 and the application slightly changed (new KafkaConnect version) and now the property will be referenced somewhere else:

    connection.user: ${file:/opt/kafka/external-configuration/connector/jdbc-details.properties:db.user}

After this update the tasks will go into error because it will say that file from version 1 (connection-details/jdbc-details.properties) cannot be found, which it's true, they should now be looking for file of version 2.
So, this leads me to believe that the tasks do not get an updated configuration.

Initially I thought the problem was the connector and the fact that my configurations were not picked up from Kafka.
Now I think the problem is with the actual tasks.

To work around this, I tried to restart the tasks with the Rest API from KafkaConnector but that did not work.

I needed to change the YAML to reduce to say 1 task instead of whatever was defined before.
Let that trigger the tasks re-creation and then increase to my initial number of tasks.
This seems to have worked, but I was wondering why this is needed.

[cstmgl]

Also trying to provide more information.
This is what I see in the connector config topics in kafka.

C:\dev>oc rsh kafka-cli-67d87d7764-jb6pp
sh-4.4$ /opt/operator/bin/kc.sh -C -t com.kafkaconnect.config -f 'topic: %t, key: %k, partition: %p, offset: %o, headers: %h\n' -e -q
topic: com.kafkaconnect.config, key: task-sink-bde-jdbc-0, partition: 0, offset: 2, headers:
...
topic: com.kafkaconnect.config, key: task-sink-bde-jdbc-47, partition: 0, offset: 49, headers:
topic: com.kafkaconnect.config, key: commit-sink-bde-jdbc, partition: 0, offset: 50, headers:
topic: com.kafkaconnect.config, key: target-state-sink-bde-avro, partition: 0, offset: 168, headers:
topic: com.kafkaconnect.config, key: task-sink-bde-avro-0, partition: 0, offset: 171, headers:
...
topic: com.kafkaconnect.config, key: task-sink-bde-avro-47, partition: 0, offset: 218, headers:
topic: com.kafkaconnect.config, key: commit-sink-bde-avro, partition: 0, offset: 219, headers:
topic: com.kafkaconnect.config, key: session-key, partition: 0, offset: 310, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-jdbc, partition: 0, offset: 311, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-avro, partition: 0, offset: 312, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-jdbc, partition: 0, offset: 313, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-tombstone-avro, partition: 0, offset: 314, headers:
topic: com.kafkaconnect.config, key: task-sink-bde-tombstone-avro-0, partition: 0, offset: 315, headers:
topic: com.kafkaconnect.config, key: commit-sink-bde-tombstone-avro, partition: 0, offset: 316, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-avro, partition: 0, offset: 317, headers:
sh-4.4$

Like I said first I thought the problem was that my configurations were stored twice in the topic and it would be a problem with log compaction

topic: com.kafkaconnect.config, key: connector-sink-bde-jdbc, partition: 0, offset: 311, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-jdbc, partition: 0, offset: 313, headers:

But now I think the problem is really that the tasks do not get updated the config somehow.
I tried restarting the tasks with the RestAPI both with just restarting the ones on error but also restarting all the ones on the connector.

curl localhost:8083
curl localhost:8083/connectors
curl -X POST localhost:8083/connectors/sink-bde-jdbc/restart?includeTasks=true
curl -X POST localhost:8083/connectors/sink-bde-avro/restart?includeTasks=true

curl localhost:8083/connectors/sink-bde-jdbc/status
curl localhost:8083/connectors/sink-bde-avro/status

After this and after a while the connector goes back to the status "RUNNING" but the tasks stayed in the restarting mode until this morning.

So this morning I tried, to modify the yaml of the KafkaConnector to define a new number of tasks (say first I put it back to 1 and then increase again)

  tasksMax: 20

Now it seems the tasks are back running, and in the configs topic I can see that it restarted

sh-4.4$ /opt/operator/bin/kc.sh -C -t com.kafkaconnect.config -f 'topic: %t, key: %k, partition: %p, offset: %o, headers: %h\n' -e -q
topic: com.kafkaconnect.config, key: task-sink-bde-jdbc-0, partition: 0, offset: 2, headers:
...
topic: com.kafkaconnect.config, key: task-sink-bde-jdbc-47, partition: 0, offset: 49, headers:
topic: com.kafkaconnect.config, key: commit-sink-bde-jdbc, partition: 0, offset: 50, headers:
topic: com.kafkaconnect.config, key: task-sink-bde-avro-0, partition: 0, offset: 171, headers:
...
topic: com.kafkaconnect.config, key: task-sink-bde-avro-47, partition: 0, offset: 218, headers:
topic: com.kafkaconnect.config, key: commit-sink-bde-avro, partition: 0, offset: 219, headers:
topic: com.kafkaconnect.config, key: session-key, partition: 0, offset: 310, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-jdbc, partition: 0, offset: 313, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-tombstone-avro, partition: 0, offset: 314, headers:
topic: com.kafkaconnect.config, key: task-sink-bde-tombstone-avro-0, partition: 0, offset: 315, headers:
topic: com.kafkaconnect.config, key: commit-sink-bde-tombstone-avro, partition: 0, offset: 316, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-avro, partition: 0, offset: 317, headers:
topic: com.kafkaconnect.config, key: restart-connector-sink-bde-jdbc, partition: 0, offset: 318, headers:
topic: com.kafkaconnect.config, key: restart-connector-sink-bde-avro, partition: 0, offset: 319, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-jdbc, partition: 0, offset: 320, headers:
topic: com.kafkaconnect.config, key: task-sink-bde-jdbc-0, partition: 0, offset: 321, headers:
topic: com.kafkaconnect.config, key: commit-sink-bde-jdbc, partition: 0, offset: 322, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-avro, partition: 0, offset: 323, headers:
topic: com.kafkaconnect.config, key: task-sink-bde-avro-0, partition: 0, offset: 324, headers:
topic: com.kafkaconnect.config, key: commit-sink-bde-avro, partition: 0, offset: 325, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-jdbc, partition: 0, offset: 326, headers:
topic: com.kafkaconnect.config, key: task-sink-bde-jdbc-0, partition: 0, offset: 327, headers:
...
topic: com.kafkaconnect.config, key: task-sink-bde-jdbc-19, partition: 0, offset: 346, headers:
topic: com.kafkaconnect.config, key: commit-sink-bde-jdbc, partition: 0, offset: 347, headers:
topic: com.kafkaconnect.config, key: connector-sink-bde-avro, partition: 0, offset: 348, headers:
topic: com.kafkaconnect.config, key: task-sink-bde-avro-0, partition: 0, offset: 349, headers:
...
topic: com.kafkaconnect.config, key: task-sink-bde-avro-19, partition: 0, offset: 368, headers:
topic: com.kafkaconnect.config, key: commit-sink-bde-avro, partition: 0, offset: 369, headers:
sh-4.4$

I was wondering if this is normal behavior in Kafka I was trying to find in the RestAPI a entry point that would allow me to reconfigure tasks instead of restarting them (because the restart did not work).

Thanks for any feedback

[scholzj]

TBH, I'm not sure I completely follow the question. The reconfiguration will normally restart the connector. It does not have any dynamic reconfiguration AFAIK.

[frankgrimes97]

FWIW, we've been bitten by this issue a number of times as well. As you said, our only known workaround when configs seem not to propagate is to resize down to 1 task then back up again.
It seems the config propagation logic is quite complex within Kafka Connect. See https://github.com/apache/kafka/blob/3.3.2/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java

From the JavaDoc:

Therefore, when reading the log, we must buffer config updates for a connector's tasks and only apply atomically them once a commit message has been read.

Perhaps the commit entry never occurs, or some Task config state remains inconsistent preventing the new config from being advertised in a new snapshot?

[frankgrimes97]

Interesting comment in this testPutTaskConfigsDoesNotResolveAllInconsistencies() test case:

Only the last call with 1 task config + 1 commit actually gets written.

https://github.com/apache/kafka/blob/b66af662e61082cb8def576ded1fe5cee37e155f/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java#L1149

[frankgrimes97]

Get the set of connectors which have inconsistent data in this snapshot. These inconsistencies can occur due to partially completed writes combined with log compaction.
Connectors in this set will appear in the output of connectors() since their connector configuration is available, but not in the output of taskConfig(ConnectorTaskId) since the task configs are incomplete.
When a worker detects a connector in this state, it should request that the connector regenerate its task configurations.

https://github.com/apache/kafka/blob/b66af662e61082cb8def576ded1fe5cee37e155f/connect/runtime/src/main/java/org/apache/kafka/connect/storage/ClusterConfigState.java#L274-L288

Odd that the only references I can find to that public Set<String> inconsistentConnectors() API call are within the unit test suite (and not from any worker-related runtime code)!

[cstmgl]

thanks for taking the time.
I still get this issue from time to time and I've been told by the colleagues at RedHat that likely this is to do with how I have my upgrade process defined. Since there is a "work-around" I have not had the time to really dive deep into this, but I fear that it may well be a bug in Kafka itself

[frankgrimes97]

FWIW, I've posted the following question on the Kafka mailing lists: https://lists.apache.org/thread/3p1b27l02zkwyzc7yn0k4wvgbco4vnof