Wrong configuration being populated to the default topicOperator

[AlphaGit]

Hi! As I'm troubleshooting a cluster that is not starting up, I found that my topics fail with this message:

org.apache.kafka.connect.errors.ConnectException: Timed out while checking for or creating topic(s) 'myservice-kafka-connect-cluster-offsets'. This could indicate a connectivity issue, unavailable topic partitions, or if this is your first use of the topic it may have taken too long to create.
	at org.apache.kafka.connect.util.TopicAdmin.createTopics(TopicAdmin.java:317)
	at org.apache.kafka.connect.storage.KafkaOffsetBackingStore$1.run(KafkaOffsetBackingStore.java:112)
	at org.apache.kafka.connect.util.KafkaBasedLog.start(KafkaBasedLog.java:130)
	at org.apache.kafka.connect.storage.KafkaOffsetBackingStore.start(KafkaOffsetBackingStore.java:128)
	at org.apache.kafka.connect.runtime.Worker.start(Worker.java:197)
	at org.apache.kafka.connect.runtime.AbstractHerder.startServices(AbstractHerder.java:128)
	at org.apache.kafka.connect.runtime.distributed.DistributedHerder.run(DistributedHerder.java:288)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=createTopics, deadlineMs=1629834651704, tries=585, nextAllowedTryMs=1629834651805) timed out at 1629834651705 after 585 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: createTopics

This lead me to troubleshoot the topics, and the topic operator itself. However, I've found mismatched configuration between the way that I've configured (and deployed) the cluster operator and what the topic operator actually has.

Here's my configuration:

apiVersion: kafka.strimzi.io/v1beta1
kind: Kafka
metadata:
  name: myservice-kafka
spec:
  kafka:
    replicas: 1
    listeners:
      - name: http
        port: 9092
        type: internal
        tls: false
    # storage redacted for simplicity
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
  zookeeper:
    replicas: 1
    # storage redacted for simplicity
  entityOperator:
    topicOperator: {} # <--- this setting is relevant
---
apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnect
metadata:
  name: myservice-kafka-connect-cluster
  annotations:
  # use-connector-resources configures this KafkaConnect
  # to use KafkaConnector resources to avoid
  # needing to call the Connect REST API directly
    strimzi.io/use-connector-resources: "true"
  labels:
    strimzi.io/cluster: myservice-kafka
spec:
  replicas: 1
  image: my-repo/kafka-connect:latest
  bootstrapServers: myservice-kafka-kafka-bootstrap.dev.svc.cluster.local:9092 # <--- this setting is relevant
  tasksMax: 1
  config:
    group.id: myservice-kafka-connect-cluster
    offset.storage.topic: myservice-kafka-connect-cluster-offsets
    config.storage.topic: myservice-kafka-connect-cluster-configs
    status.storage.topic: myservice-kafka-connect-cluster-status
    key.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: false
    value.converter.schemas.enable: false
    config.storage.replication.factor: 1
    offset.storage.replication.factor: 1
    status.storage.replication.factor: 1
  # resources redacted for simplicity
  template:
    pod:
      imagePullSecrets:
        - name: mySecret

However, when I review the topic operator logs, I find that it is using a different configuration:

kubectl -n dev logs myservice-kafka-entity-operator-bb47d9dff-xsxlb topic-operator | grep -A30 "Using config" 

021-08-24 21:00:39,54658 INFO  [main] Session:72 - Using config:
	STRIMZI_TRUSTSTORE_LOCATION: /tmp/topic-operator/replication.truststore.p12
	STRIMZI_RESOURCE_LABELS: strimzi.io/cluster=myservice-kafka
	STRIMZI_FULL_RECONCILIATION_INTERVAL_MS: 90000
	STRIMZI_CLIENT_ID: strimzi-topic-operator-46af93d4-8d00-4a3c-954b-8d782f495a1e
	STRIMZI_STALE_RESULT_TIMEOUT_MS: 5000
	STRIMZI_TOPIC_METADATA_MAX_ATTEMPTS: 6
	STRIMZI_KEYSTORE_LOCATION: /tmp/topic-operator/replication.keystore.p12
	STRIMZI_REASSIGN_THROTTLE: 9223372036854775807
	STRIMZI_USE_ZOOKEEPER_TOPIC_STORE: false
	STRIMZI_KAFKA_BOOTSTRAP_SERVERS: myservice-kafka-kafka-bootstrap:9091 # < --- why is this different?
	STRIMZI_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM: HTTPS
	STRIMZI_NAMESPACE: dev
	STRIMZI_APPLICATION_ID: __strimzi-topic-operator-kstreams
	STRIMZI_APPLICATION_SERVER: localhost:9000
	STRIMZI_ZOOKEEPER_SESSION_TIMEOUT_MS: 20000
	STRIMZI_DISTRIBUTED_STORE: false
	STRIMZI_TOPICS_PATH: /strimzi/topics
	STRIMZI_ZOOKEEPER_CONNECT: localhost:2181  # < --- why is this set to localhost, different from the cluster svc address?
	STRIMZI_TLS_ENABLED: true
	STRIMZI_KEYSTORE_PASSWORD: ********
	STRIMZI_STORE_NAME: topic-store
	STRIMZI_REASSIGN_VERIFY_INTERVAL_MS: 120000
	TC_ZK_CONNECTION_TIMEOUT_MS: 20000
	STRIMZI_TRUSTSTORE_PASSWORD: ********
	STRIMZI_STORE_TOPIC: __strimzi_store_topic

So, my hypothesis is that the topic operator cannot connect to the zookeper/boostraping servers and that's why the topic creation is failing. Am I on the right track? How do I set the right properties?

---

Update: Adding the kafka pod logs and the kafka connect pod logs.

• connect-logs.log
• kafka-logs.log
• full-kafka-logs.log

[scholzj]

You hypothesis is wrong:

• The Topic Operator connects to Strimzi-internal Kafka listener on port 9091 (it cannot rely on the listener on port 9092 which you configure and which might not even exist or might have completely different configuration). It connects to ZooKeeper using TLS sidecar listening on localhost:2181.
• But more important, the Connect error has nothing to do with the Topic Operator. Connect creates the topic directly in Kafka, it doesn't really care about the Topic Operator one way or another.

The Connect error you have suggests that it is either having troubles connecting to the Kafka broker or trouble creating the topic. Without the full logs from both, it is not easy to say what could be the issue. I think the configuration looks good.

[scholzj]

The Kafka log does not seem to start from beginning, so I'm not sure if it maybe missed something. Are these both in the same namespace? If not, it might be good to try to probe the Kafka broker with telnet or something from the other namespace to see if the connectivity works on the networking level.

[AlphaGit]

I have updated the question again to include the full kafka logs, sorry about that. Yes, these are both in the same namespace. I appreciate your help in trying to figure this out. :)

[scholzj]

Hmm, the full log shows that the topic is created and seems to be online. So TBH I have no idea about what could be the issue. Have you tried the telnet connection?

[AlphaGit]

I'll try that out and get back to you. (Just letting you know since it might take me a bit.)

[AlphaGit]

Hi! I apologize I never came back to this. The problem eventually solved itself between re-deployments of the helm chart that contained kafka, but I think I was able to put together what happened initially that brought me here.

At some point, in my cluster, the kafka cluster instances did not have a place to be (as there were no free nodes with the requirements fulfilled), which let the cluster not being able to start at all.

While troubleshooting this, we completely removed kafka and re-created it, but along with doing this, the offset topic was manually removed. When re-deploying, it wasn't manually re-created. As such, Kafka was failing to use or create the offset topic (we expected it to create it if it didn't find it). At this point we saw the networking issues we found here.

Fully removing the three topics (config, offset, status) and redeploying kafka ensuring that all instances had a place to run did autocreate the three topics again, and the topic operator did report success in communicating to them.