How to deploy Kafka Connect in distributed mode across multiple K8S clusters

[jooomOut]

Hello.

I deploy Kafka Connect using Strimzi with the same configuration across more than two K8S clusters.
I want to logically use these multiple K8S clusters as a single Kafka Connect cluster by using the same group.id and config topics. (distributed-mode)

However, when the Kafka Connect cluster is configured in this way, workers cannot find the leader worker, making the REST API unusable. I suspect this is because the rest.advertised.host.name is set to the worker's name, which prevents communication to workers in other K8S clusters. If, by chance, API requests reach the K8S cluster where the leader worker is located, the API functions properly. However, even in this situation, communication between workers fails, preventing task creation.

Therefore, I thought that changing the rest.advertised.host.name to the host IP and exposing it via nodeport might solve the issue of enabling communication between worker nodes located in different K8S clusters.
However, I found that in the Strimzi configuration, the rest.* options are rejected.

Question

Does Strimzi not support using multiple K8S clusters as a single Kafka Connect cluster in this manner?
How can I use multiple K8S clusters as a single Kafka Connect cluster?

My Config

Below is my Strimzi K8S config. It is being deployed identically on two K8S clusters. (excluding the ingress-related configurations)

• strimzi : 0.43.0
• kafka connect : 3.8.0

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: src-ct
  labels:
    k8s-app: src-ct
    app: src-ct
spec:
  version: 3.8.0
  replicas: {{ .Values.kafkaConnect.replicas }}
  bootstrapServers: {{ .Values.kafkaConnect.bootstrapServers }}
  image: {{ .Values.kafkaConnect.image }}
  config:
    group.id: {{ .Values.kafkaConnect.groupId }}
    offset.storage.topic: {{ .Values.kafkaConnect.storageTopic }}-offsets
    config.storage.topic: {{ .Values.kafkaConnect.storageTopic }}-configs
    status.storage.topic: {{ .Values.kafkaConnect.storageTopic }}-statuses
    config.storage.replication.factor: -1
    offset.storage.replication.factor: -1
    status.storage.replication.factor: -1
    key.converter: io.confluent.connect.avro.AvroConverter
    key.converter.schema.registry.url: {{ .Values.schemaRegistry}}
    value.converter: io.confluent.connect.avro.AvroConverter
    value.converter.schema.registry.url: {{ .Values.schemaRegistry}}
    producer.max.request.size: 10485760 # 10MB
  logging:
    type: inline
    loggers:
      log4j.rootLogger: {{ .Values.kafkaConnect.logging.rootLogger }}
      log4j.logger.io.debezium.connector.jdbc: {{ .Values.kafkaConnect.logging.jdbcLogger }}
      log4j.logger.io.debezium.connector.mysql: {{ .Values.kafkaConnect.logging.mysqlLogger }}
      log4j.logger.org.apache.kafka.connect.runtime.rest.RestServer: {{ .Values.kafkaConnect.logging.restServerLogger }}
  jvmOptions:
    javaSystemProperties:
      - name: file.encoding
        value: "UTF-8"
      - name: networkaddress.cache.ttl
        value: "0"
      - name: networkaddress.cache.negative.ttl
        value: "0"
  template:
    pod:
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
            - weight: 10
              podAffinityTerm:
                labelSelector:
                  matchExpressions:
                    - key: app
                      operator: In
                      values:
                        - src-ct
                topologyKey: "kubernetes.io/hostname"

---
apiVersion: v1
kind: Service
metadata:
  name: src-ct-lb
  labels:
    app: kafka-connect
spec:
  type: LoadBalancer
  ports:
  - name: external
    port: 8083
    targetPort: 8083
    protocol: TCP
  selector:
    strimzi.io/cluster: src-ct
    strimzi.io/kind: KafkaConnect

log

2025-02-06 09:53:06,733 ERROR Request 'POST /connectors/test/restart' failed because it couldn't find the target Connect worker within two hops (between workers). (org.apache.kafka.connect.runtime.rest.HerderRequestHandler) [qtp1966930659-230]
2025-02-06 09:53:06,963 ERROR Request 'POST /connectors/test/restart' failed because it couldn't find the target Connect worker within two hops (between workers). (org.apache.kafka.connect.runtime.rest.HerderRequestHandler) [qtp1966930659-216]
2025-02-06 09:53:07,198 ERROR Error forwarding REST request (org.apache.kafka.connect.runtime.rest.RestClient) [qtp1966930659-230]
org.apache.kafka.connect.runtime.rest.errors.ConnectRestException: Cannot complete request because of a conflicting operation (e.g. worker rebalance)
	at org.apache.kafka.connect.runtime.rest.RestClient.httpRequest(RestClient.java:183)
	at org.apache.kafka.connect.runtime.rest.RestClient.httpRequest(RestClient.java:137)
	at org.apache.kafka.connect.runtime.rest.RestClient.httpRequest(RestClient.java:81)
	at org.apache.kafka.connect.runtime.rest.HerderRequestHandler.completeOrForwardRequest(HerderRequestHandler.java:124)
	at org.apache.kafka.connect.runtime.rest.HerderRequestHandler.completeOrForwardRequest(HerderRequestHandler.java:141)
	at org.apache.kafka.connect.runtime.rest.HerderRequestHandler.completeOrForwardRequest(HerderRequestHandler.java:151)
	at org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource.restartConnector(ConnectorsResource.java:272)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:569)
	at org.glassfish.jersey.server.model.internal.ResourceMethodInvocationHandlerFactory.lambda$static$0(ResourceMethodInvocationHandlerFactory.java:52)
	at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher$1.run(AbstractJavaResourceMethodDispatcher.java:134)
	at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.invoke(AbstractJavaResourceMethodDispatcher.java:177)
	at org.glassfish.jersey.server.model.internal.JavaResourceMethodDispatcherProvider$ResponseOutInvoker.doDispatch(JavaResourceMethodDispatcherProvider.java:176)
	at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.dispatch(AbstractJavaResourceMethodDispatcher.java:81)
	at org.glassfish.jersey.server.model.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:478)
	at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:400)
	at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:81)
	at org.glassfish.jersey.server.ServerRuntime$1.run(ServerRuntime.java:256)
	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:248)
	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:244)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:292)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:274)
	at org.glassfish.jersey.internal.Errors.process(Errors.java:244)
	at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:265)
	at org.glassfish.jersey.server.ServerRuntime.process(ServerRuntime.java:235)
	at org.glassfish.jersey.server.ApplicationHandler.handle(ApplicationHandler.java:684)
	at org.glassfish.jersey.servlet.WebComponent.serviceImpl(WebComponent.java:394)
	at org.glassfish.jersey.servlet.WebComponent.service(WebComponent.java:346)
	at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:358)
	at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:311)
	at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:205)
	at org.eclipse.jetty.servlet.ServletHolder.handle(ServletHolder.java:799)
	at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:554)
	at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233)
	at org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:1624)
	at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233)
	at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1440)
	at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:188)
	at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:505)
	at org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:1594)
	at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:186)
	at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1355)
	at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141)
	at org.eclipse.jetty.server.handler.ContextHandlerCollection.handle(ContextHandlerCollection.java:191)
	at org.eclipse.jetty.server.handler.StatisticsHandler.handle(StatisticsHandler.java:181)
	at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127)
	at org.eclipse.jetty.server.Server.handle(Server.java:516)
	at org.eclipse.jetty.server.HttpChannel.lambda$handle$1(HttpChannel.java:487)
	at org.eclipse.jetty.server.HttpChannel.dispatch(HttpChannel.java:732)
	at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:479)
	at org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:277)
	at org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:311)
	at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105)
	at org.eclipse.jetty.io.ChannelEndPoint$1.run(ChannelEndPoint.java:104)
	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.runTask(EatWhatYouKill.java:338)
	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.doProduce(EatWhatYouKill.java:315)
	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.tryProduce(EatWhatYouKill.java:173)
	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.run(EatWhatYouKill.java:131)
	at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.run(ReservedThreadExecutor.java:409)
	at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:883)
	at org.eclipse.jetty.util.thread.QueuedThreadPool$Runner.run(QueuedThreadPool.java:1034)
	at java.base/java.lang.Thread.run(Thread.java:840)

Thanks.

[scholzj]

That is not supported and will not work currently.