Facing issues while trying to run snowflake kafka connector using strimzi kafka connect CR

[piyushbhojgude]

Hi,

I'm trying to setup a consumer using strimzi kafka connect CR and use snowflake kafka connector.
And I'm seeing two issues while doing so.

1. One is the kafka connect pod is taking a long time to come up in ready state.
2. The snowflake kafka connector is failing to start.

Here is my Dockerfile definition:

# Use Strimzi kafka image for kafka connect
FROM quay.io/strimzi/kafka:0.37.0-kafka-3.5.1-amd64

USER root:root

RUN mkdir -p /opt/kafka/plugins/snowflake

RUN curl -sSL "https://repo1.maven.org/maven2/com/snowflake/snowflake-kafka-connector/2.1.0/snowflake-kafka-connector-2.1.0.jar" -o  /opt/kafka/plugins/snowflake/snowflake-kafka-connector-2.1.0.jar
RUN curl -sSL "https://repo1.maven.org/maven2/net/snowflake/snowflake-ingest-sdk/2.0.2/snowflake-ingest-sdk-2.0.2.jar" -o /opt/kafka/plugins/snowflake/snowflake-ingest-sdk-2.0.2.jar
RUN curl -sSL "https://repo1.maven.org/maven2/net/snowflake/snowflake-jdbc/3.13.15/snowflake-jdbc-3.13.15.jar" -o /opt/kafka/plugins/snowflake/snowflake-jdbc-3.13.15.jar
RUN curl -sSL "https://repo1.maven.org/maven2/org/bouncycastle/bc-fips/1.0.1/bc-fips-1.0.1.jar" -o /opt/kafka/plugins/snowflake/bc-fips-1.0.1.jar
RUN curl -sSL "https://repo1.maven.org/maven2/org/bouncycastle/bcpkix-fips/1.0.3/bcpkix-fips-1.0.3.jar" -o /opt/kafka/plugins/snowflake/bcpkix-fips-1.0.3.jar

RUN cp /opt/kafka/plugins/snowflake/snowflake-kafka-connector-2.1.0.jar /opt/kafka/libs/
RUN cp /opt/kafka/plugins/snowflake/bcpkix-fips-1.0.3.jar /opt/kafka/libs/
RUN cp /opt/kafka/plugins/snowflake/bc-fips-1.0.1.jar /opt/kafka/libs/

USER 1001

I'm building, tagging and pushing this dockerfile to AWS ECR, since my Kafka cluster is deployed on EKS.
Here is my Kafka connect yaml def:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: strimzi-snowflake-kafka
  namespace: "kafka-servers"
  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"
spec:
  replicas: 1
  bootstrapServers: kafka-cluster-snowflake-kafka-bootstrap.kafka-servers.svc:9092
  image: {ecr-url}/strimzi-snowflake-kafka-connect:latest
  config:
    config.providers: env
    config.providers.env.class: org.apache.kafka.common.config.provider.EnvVarConfigProvider
    group.id: strimzi-kafka-connect-cluster
    offset.flush.interval.ms: 10000
    offset.storage.topic: strimzi-kafka-connect-offsets
    config.storage.topic: strimzi-kafka-connect-configs
    status.storage.topic: strimzi-kafka-connect-status
    key.converter: org.apache.kafka.connect.storage.StringConverter
    value.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: false
    value.converter.schemas.enable: false
  readinessProbe:
    initialDelaySeconds: 1000
    timeoutSeconds: 1000
  livenessProbe:
      initialDelaySeconds: 1200
      timeoutSeconds: 1200
  externalConfiguration:
    env:
      - name: SNOWFLAKE_USER_NAME
        valueFrom:
          secretKeyRef:
            name: snowflake-secret
            key: userName
      - name: SNOWFLAKE_PRIVATE_KEY
        valueFrom:
          secretKeyRef:
            name: snowflake-secret
            key: privateKey
      - name: SNOWFLAKE_PRIVATE_KEY_PASSPHRASE
        valueFrom:
          secretKeyRef:
            name: snowflake-secret
            key: privateKeyPassphrase
      - name: SNOWFLAKE_DATABASE_NAME
        valueFrom:
          secretKeyRef:
            name: snowflake-secret
            key: databaseName
      - name: SNOWFLAKE_SCHEMA_NAME
        valueFrom:
          secretKeyRef:
            name: snowflake-secret
            key: schemaName
      - name: SNOWFLAKE_ROLE_NAME
        valueFrom:
          secretKeyRef:
            name: snowflake-secret
            key: roleName
      - name: SNOWFLAKE_URL_NAME
        valueFrom:
          secretKeyRef:
            name: snowflake-secret
            key: snowflakeUrlName

Currently for testing I'm deploying this using kubectl apply -f strimzi-kafka-connect.yaml -n kafka-servers

Post deploying I can see that the pod is running but not in ready state. Sometimes it takes 5 mins to get into ready state whereas sometimes it never gets into ready state. So not sure what is going wrong in this case. How do I go about debugging this?

Piyush.Bhojgude@ip-192-168-1-3 temp % kubectl get pods -n kafka-servers | grep strimzi-snowflake
strimzi-snowflake-kafka-connect-dcdf8fb9d-s7qk7           0/1     Running   0          9m19s

Here are the last few logs for the pod:

2024-03-13 04:25:36,680 INFO Started o.e.j.s.ServletContextHandler@56b704ea{/,null,AVAILABLE} (org.eclipse.jetty.server.handler.ContextHandler) [main]
2024-03-13 04:25:36,680 INFO REST resources initialized; server is started and ready to handle requests (org.apache.kafka.connect.runtime.rest.RestServer) [main]
2024-03-13 04:25:36,680 INFO Kafka Connect started (org.apache.kafka.connect.runtime.Connect) [main]
2024-03-13 04:25:39,316 INFO [Worker clientId=connect-1, groupId=strimzi-kafka-connect-cluster] Successfully joined group with generation Generation{generationId=5, memberId='connect-1-a413794e-01a8-4d0a-bcb1-dbdf43560996', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2024-03-13 04:25:39,336 INFO [Worker clientId=connect-1, groupId=strimzi-kafka-connect-cluster] Successfully synced group in generation Generation{generationId=5, memberId='connect-1-a413794e-01a8-4d0a-bcb1-dbdf43560996', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2024-03-13 04:25:39,336 INFO [Worker clientId=connect-1, groupId=strimzi-kafka-connect-cluster] Joined group at generation 5 with protocol version 2 and got assignment: Assignment{error=0, leader='connect-1-a413794e-01a8-4d0a-bcb1-dbdf43560996', leaderUrl='http://100.64.133.234:8083/', offset=9, connectorIds=[], taskIds=[], revokedConnectorIds=[], revokedTaskIds=[], delay=0} with rebalance delay: 0 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2024-03-13 04:25:39,336 WARN [Worker clientId=connect-1, groupId=strimzi-kafka-connect-cluster] Catching up to assignment's config offset. (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2024-03-13 04:25:39,336 INFO [Worker clientId=connect-1, groupId=strimzi-kafka-connect-cluster] Current config state offset -1 is behind group assignment 9, reading to end of config log (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2024-03-13 04:25:39,369 INFO [Worker clientId=connect-1, groupId=strimzi-kafka-connect-cluster] Finished reading to end of log and updated config snapshot, new config log offset: 9 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2024-03-13 04:25:39,369 INFO [Worker clientId=connect-1, groupId=strimzi-kafka-connect-cluster] Starting connectors and tasks using config offset 9 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2024-03-13 04:25:39,369 INFO [Worker clientId=connect-1, groupId=strimzi-kafka-connect-cluster] Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]

Once the strimzi-kafka-connect pod is in ready state, I'm applying the snowflake-kafka-connector.yaml.
Here is the def for snowflake-kafka-connector.yaml:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: snowflake-connector-int
  namespace: "kafka-servers"
  labels:
    strimzi.io/cluster: strimzi-snowflake-kafka
spec:
  class: com.snowflake.kafka.connector.SnowflakeSinkConnector
  tasksMax: 1
  autoRestart:
    enabled: true
  config:
    name: "snowflake-connector-int"
    topics: "cloudtrail-logs"
    snowflake.topic2table.map: "cloudtrail-logs:connect_test"
    buffer.count.records: "10000"
    buffer.flush.time: "10"
    buffer.size.bytes: "20000000"
    snowflake.url.name: ${env:SNOWFLAKE_URL_NAME}
    snowflake.user.name: ${env:SNOWFLAKE_USER_NAME}
    snowflake.private.key: ${env:SNOWFLAKE_PRIVATE_KEY}
    snowflake.private.key.passphrase: ${env:SNOWFLAKE_PRIVATE_KEY_PASSPHRASE}
    snowflake.database.name: ${env:SNOWFLAKE_DATABASE_NAME}
    snowflake.schema.name: ${env:SNOWFLAKE_SCHEMA_NAME}
    snowflake.role.name: ${env:SNOWFLAKE_ROLE_NAME}
    snowflake.ingestion.method: SNOWPIPE_STREAMING
    snowflake.enable.schematization: "FALSE"
    key.converter: org.apache.kafka.connect.storage.StringConverter
    value.converter: org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable: false
    value.converter.schemas.enable: false
    behavior.on.null.values: IGNORE
    enable.streaming.client.optimization: TRUE

But the connector fails to start with following error:

2024-03-13 03:23:41,261 INFO [AdminClient clientId=strimzi-kafka-connect-cluster--shared-admin] Node 0 disconnected. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | strimzi-kafka-connect-cluster--shared-admin]
2024-03-13 03:23:44,893 INFO 100.64.162.25 - - [13/Mar/2024:03:23:44 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.28+" 5 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-22]
2024-03-13 03:23:44,898 INFO 100.64.162.25 - - [13/Mar/2024:03:23:44 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.28+" 10 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-26]
2024-03-13 03:23:54,945 INFO 100.64.162.25 - - [13/Mar/2024:03:23:54 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.28+" 6 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-22]
2024-03-13 03:23:54,948 INFO 100.64.162.25 - - [13/Mar/2024:03:23:54 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.28+" 15 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-25]
2024-03-13 03:24:04,893 INFO 100.64.162.25 - - [13/Mar/2024:03:24:04 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.28+" 4 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-24]
2024-03-13 03:24:04,896 INFO 100.64.162.25 - - [13/Mar/2024:03:24:04 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.28+" 8 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-22]
2024-03-13 03:24:14,871 INFO 100.64.188.101 - - [13/Mar/2024:03:24:14 +0000] "GET /admin/loggers/ HTTP/1.1" 200 142 "-" "-" 6 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-24]
2024-03-13 03:24:14,873 INFO 100.64.188.101 - - [13/Mar/2024:03:24:14 +0000] "GET /connectors HTTP/1.1" 200 2 "-" "-" 8 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-20]
2024-03-13 03:24:14,874 INFO 100.64.188.101 - - [13/Mar/2024:03:24:14 +0000] "GET /connector-plugins HTTP/1.1" 200 405 "-" "-" 10 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-22]
2024-03-13 03:24:14,892 INFO 100.64.162.25 - - [13/Mar/2024:03:24:14 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.28+" 5 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-27]
2024-03-13 03:24:14,892 INFO 100.64.162.25 - - [13/Mar/2024:03:24:14 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.28+" 5 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-23]
2024-03-13 03:24:24,891 INFO 100.64.162.25 - - [13/Mar/2024:03:24:24 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.28+" 3 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-26]
2024-03-13 03:24:24,892 INFO 100.64.162.25 - - [13/Mar/2024:03:24:24 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.28+" 4 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-27]
2024-03-13 03:24:34,894 INFO 100.64.162.25 - - [13/Mar/2024:03:24:34 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.28+" 6 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-25]
2024-03-13 03:24:34,900 INFO 100.64.162.25 - - [13/Mar/2024:03:24:34 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.28+" 10 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-26]
2024-03-13 03:24:39,621 INFO 100.64.188.101 - - [13/Mar/2024:03:24:39 +0000] "GET /connectors/snowflake-connector-int/config HTTP/1.1" 404 74 "-" "-" 9 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp639224694-22]
2024-03-13 03:24:39,859 WARN /connectors/snowflake-connector-int/config (org.eclipse.jetty.server.HttpChannel) [qtp639224694-25]
javax.servlet.ServletException: org.glassfish.jersey.server.ContainerException: java.lang.NoClassDefFoundError: org/bouncycastle/jcajce/provider/BouncyCastleFipsProvider
	at org.glassfish.jersey.servlet.WebComponent.serviceImpl(WebComponent.java:410)
	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:833)

I am adding the bc-fips-1.0.1.jar and also the bcpkix-fips-1.0.3.jar as part of my container creation. So again not sure why its not able to find the Class def.

[scholzj]

Not really sure why are you putting the JARs into the libs directory. You should have it just in the /opt/kafka/plugins/snowflake/. You should also keep in mind that these errors are a bit tricky. It sometimes means that some other JARs with classes used by this class are missing etc. So you should make sure the JARs you add there are really all that is needed.

[piyushbhojgude]

Got you, I went through the snowflake kafka connector doc: https://docs.snowflake.com/en/user-guide/kafka-connector-install#installing-the-connector-for-confluent
Based on this I'm now just adding snowflake-kafka-connector jar along with bc-fips and bcpkix-fips in the plugins folder. Not pushing it to libs folder.
And now the connector seems to be running fine 👍

I had couple of more queries @scholzj ,
Suppose I'm creating a kafka-connect-cluster with 3 nodes. While creating the cluster I configure the spec as follows:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: strimzi-sf-kafka
  namespace: "kafka-servers"
  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"
spec:
  replicas: 3
  bootstrapServers: kafka-cluster-snowflake-kafka-bootstrap.kafka-servers.svc:9092
  image: {ecr-url}/strimzi-snowflake-kafka-connect:latest
  config:
    config.providers: env
    config.providers.env.class: org.apache.kafka.common.config.provider.EnvVarConfigProvider
    group.id: strimzi-kafka-connect-cluster
    offset.flush.interval.ms: 10000
    offset.storage.topic: strimzi-kafka-connect-offsets
    config.storage.topic: strimzi-kafka-connect-configs
    status.storage.topic: strimzi-kafka-connect-status

Now if I want to deploy 2 different connectors, I would create two separate kafkaConnector deployments.
And if I mention the same connect cluster name strimzi-sf-kafka for both, would this imply that both connectors would run in the same kafka connect cluster?
Also would the two connectors use the same offset.storage.topic, config.storage.topic and status.storage.topic ?
Furthermore if I have 2 tasks for 1st connector and 4 tasks for 2nd connector, I'm assuming all these tasks would be distributed across the 3 worker nodes under one kafka connect cluster.

So if I understand this correctly, the group.id variable in the kafka-connect spec is for worker group and not for the consumer group right?
If I deploy 2 connectors with different names, then those would be deployed as 2 separate consumer groups, First one with 2 tasks and second one with 4 tasks.

And in case if I want to shift the 2nd connector to new kafka-connect-cluster, I would need to deploy a new kafka-connect cluster and then deploy the 2nd connector against the same.
Is this understanding also correct ?

[scholzj]

Now if I want to deploy 2 different connectors, I would create two separate kafkaConnector deployments.
And if I mention the same connect cluster name strimzi-sf-kafka for both, would this imply that both connectors would run in the same kafka connect cluster?

Yes, this would schedule the connectors on the same Connect cluster. Unless there is something special with the Snowflake connector, this normally works fine. But keep in mind that you should not expect any strong security sandboxing between the connectors, they will run in the same JVM etc. There is also no isolation in terms of how many resources will each connector get.

Also would the two connectors use the same offset.storage.topic, config.storage.topic and status.storage.topic ?

These topics belong to the Connect cluster. As far as I know, they are not used directly by the connectors. But yes, they would be shared among all the connectors in given Connect cluster.

Furthermore if I have 2 tasks for 1st connector and 4 tasks for 2nd connector, I'm assuming all these tasks would be distributed across the 3 worker nodes under one kafka connect cluster.

Yes, they will be distributed. But keep in mind that there is no sophisticated algorithm for how they are scheduled. It should work fine assuming all the tasks are similar in size (of the resources they need). If you have some tasks that are super busy and need a lot of resources and some that are only small, the scheduling of the tasks might be suboptimal.

So if I understand this correctly, the group.id variable in the kafka-connect spec is for worker group and not for the consumer group right?
If I deploy 2 connectors with different names, then those would be deployed as 2 separate consumer groups, First one with 2 tasks and second one with 4 tasks.

Yes, the group.id from the KafkaConnect resource is the one used by Connect itself. The connectors are independent on each other. Their group IDs wll normally be something like connect-<connector-name>.

And in case if I want to shift the 2nd connector to new kafka-connect-cluster, I would need to deploy a new kafka-connect cluster and then deploy the 2nd connector against the same.
Is this understanding also correct ?

I have never done it. So I'm not 100% sure. If you are consuming from Kafka (i.e. have a sink connector), I think it uses consumer groups to store its progress on Kafka side. That will survive independently on the Connect cluster. But everything stored in the COnnect topics will not be present on the other Connect cluster. So it will depend on the actual connector what will the impact be when you try to move it to another Connect cluster.

[piyushbhojgude]

Thank you for the prompt response @scholzj .