Meshed application is unable to connect to a Strimzi operator-managed Kafka cluster

[UncleIS]

Hi, everyone!

We are currently experimenting with Linkerd service mesh. One of the scenarios that we are testing is the ability of a Java SpringBoot application (single pod) to talk to a Kafka cluster (single broker node) via Linkerd: that would allow us to encrypt Kafka traffic without the hassle of managing certificates, etc. Our setup is as follows:

We run a self-hosted Rancher RKE2 Kubernetes cluster (version v1.23.14+rke2r1) on a bunch of VMs in a private cloud (Oracle Linux 8 is the underlying OS).
Java SpringBoot application runs in dev namespace of the cluster. It manages two containers: one for the application, another - for Hashicorp Vault agent (alongside with Vault Agent init container):

$ kubectl -n dev get pods app-78d45f67c8-mcslb 
NAME                     READY   STATUS    RESTARTS   AGE
app-78d45f67c8-mcslb     2/2     Running   0          25m

Strimzi-operated Kafka cluster runs in strimzi namespace of the same K8S cluster (operator version 0.32.0):

$ helm -n strimzi list
NAME            NAMESPACE       REVISION        UPDATED                                 STATUS          CHART                           APP VERSION     
strimzi-kafka   strimzi         3               2022-11-29 13:26:14.798434407 +0000 UTC deployed        strimzi-kafka-operator-0.32.0   0.32.0

$ kubectl -n strimzi get kafka kafka-cluster 
NAME                 DESIRED KAFKA REPLICAS   DESIRED ZK REPLICAS   READY   WARNINGS
kafka-cluster        1                        1                     True    True

$ kubectl -n strimzi get pods
NAME                                                 READY   STATUS    RESTARTS      AGE
kafka-cluster-entity-operator-95dfc5fb6-lzzfk        3/3     Running   9 (44h ago)   5d1h
kafka-cluster-kafka-0                                1/1     Running   0             42m
kafka-cluster-kafka-exporter-758fd6d4ff-zdqtf        1/1     Running   0             5d1h
kafka-cluster-zookeeper-0                            1/1     Running   0             44h
strimzi-cluster-operator-6f9886f64d-f999r            2/2     Running   7 (44h ago)   46h 

It is worth noting that kafka-cluster-kafka-0 pod along with the other pods is not managed by a StatefulSet. Instead, it is managed by a StrimziPodSet:

$ kubectl -n strimzi get sts
No resources found in strimzi namespace.

$ kubectl -n strimzi get strimzipodsets.core.strimzi.io 
NAME                           PODS   READY PODS   CURRENT PODS   AGE
kafka-cluster-kafka	       	1      1            1              223d
kafka-cluster-zookeeper		1      1            1              223d

The Kafka broker exposes port 9092 (among others) as a plain internal port (no TLS / authentication enabled) via a number of services, both ClusterIP and Headless:

$ kubectl -n strimzi get svc
NAME                                          TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)                                        AGE
kafka-cluster-kafka-0                    ClusterIP   10.43.253.58    <none>        9094/TCP                                       223d
kafka-cluster-kafka-bootstrap            ClusterIP   10.43.128.131   <none>        9091/TCP,9092/TCP,9093/TCP            433d
kafka-cluster-kafka-brokers              ClusterIP   None            <none>        9090/TCP,9091/TCP,9092/TCP,9093/TCP   433d
kafka-cluster-kafka-external-bootstrap   ClusterIP   10.43.36.66     <none>        9094/TCP                                       223d
kafka-cluster-zookeeper-client           ClusterIP   10.43.229.175   <none>        2181/TCP                                       433d
kafka-cluster-zookeeper-nodes            ClusterIP   None            <none>        2181/TCP,2888/TCP,3888/TCP                     433d

The application is configured to connect to the cluster via a ConfigMap setting an environment variable:

SPRING_KAFKA_BOOTSTRAP_SERVERS: kafka-cluster-kafka-brokers.strimzi.svc.cluster.local:9092

Without Linkerd, this setup works fine: the application is able to connect broker node and interact with its topics. Similarly, if we have Linkerd injected on either side of this setup, the application is able to connect without issues.

For Kafka, we inject Linkerd via the annotations of Kafka custom resource:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
spec:
  kafka:
    template:
      bootstrapService:
        metadata:
          annotations:
            config.linkerd.io/opaque-ports: 9092,9093,9094
      brokersService:
        metadata:
          annotations:
            config.linkerd.io/opaque-ports: 9092,9093,9094
      externalBootstrapService:
        metadata:
          annotations:
            config.linkerd.io/opaque-ports: 9092,9093,9094
      perPodService:
        metadata:
          annotations:
            config.linkerd.io/opaque-ports: 9092,9093,9094
      pod:
        metadata:
          annotations:
            config.linkerd.io/opaque-ports: 9092,9093,9094
            config.linkerd.io/proxy-log-level: debug
            config.linkerd.io/skip-inbound-ports: 9090,9091
            linkerd.io/inject: enabled

Injection of Linkerd to the application pod is done via the kubectl -n dev get deploy app | linkerd inject - | kubectl apply -f - command pipe.

Once we have Linkerd injected on both sides of the setup, the application is no longer able to connect to Kafka, as the connection times out. In application logs, we see:

{"timestamp":"2023-06-07 08:14:12.007","level":"WARN","thread":"org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1","logger":"org.apache.kafka.clients.NetworkClient","message":"[Consumer clientId=consumer-app-3, groupId=app] Bootstrap broker kafka-cluster-kafka-brokers.strimzi.svc.cluster.local:9092 (id: -1 rack: null) disconnected","context":"default"}

In logs of the proxy on the application side, there are records as follows:

[    74.429343s] DEBUG ThreadId(01) outbound:accept{client.addr=10.42.9.214:39970}:proxy{addr=10.42.17.189:9092}: tower::buffer::worker: service.ready=true processing request
[    74.429433s] DEBUG ThreadId(01) outbound:accept{client.addr=10.42.9.214:39970}:proxy{addr=10.42.17.189:9092}: tower::buffer::worker: service.ready=true processing request
[    74.429459s] DEBUG ThreadId(01) outbound:accept{client.addr=10.42.9.214:39970}:proxy{addr=10.42.17.189:9092}:forward{addr=10.42.17.189:9092}: linkerd_app_outbound::tcp::tagged_transport: Using opaque transport target_port=9092 opaque_port=4143
[    74.429474s] DEBUG ThreadId(01) outbound:accept{client.addr=10.42.9.214:39970}:proxy{addr=10.42.17.189:9092}:forward{addr=10.42.17.189:9092}: linkerd_proxy_transport::connect: Connecting server.addr=10.42.17.189:4143
[    75.430331s]  INFO ThreadId(01) outbound:accept{client.addr=10.42.9.214:39970}: linkerd_app_core::serve: Connection closed error=connect timed out after 1s client.addr=10.42.9.214:39970

In the debug logs of the Kafka Pod, there are no traces of the connection attempts from this pod.

The timeout of 1s is coming from the proxy: we experimented with increasing it to up to 1m via annotations, but this didn't make any difference.

So far we have tried to use other services' DNS names and even the IP address of the Kafka broker Pod in SPRING_KAFKA_BOOTSTRAP_SERVERS environemnt variable, but non fixed the issue.

Any help would be much appreciated.

[andrewortman]

Hey, its been a long time since you asked this, but did you by chance resolve this issue? I'm seeing this exact problem in 2.19.

[bdols]

linkerd probably doesn't handle kafka protocol's idiosyncracies, see https://linkerd.io/2-edge/features/protocol-detection/#protocols-that-may-require-configuration for latest. If it helps: I am using strimzi's TLS client certs that strimzi's operator provisions and rotates instead of meshing, so you would need to provide strimzi's trust store and client certs to your application, of course.