Snowflake Connector Plugin No Errors and No Data

[rick-reihl]

Hi I'm working on setting up a snowflake sink connector to strimzi and have gotten stuck and need some help. I have mainly used this post as guidance: https://github.com/orgs/strimzi/discussions/9825.

I have applied these connector configs. I am able to produce using the produce.py script and consume with an equivalent one, but no data is going to snowflake, and notably kubectl get kafkaconnector yields the following where READY is blank.

NAME                  CLUSTER   CONNECTOR CLASS                                        MAX TASKS   READY
snowflake-connector             com.snowflake.kafka.connector.SnowflakeSinkConnector   8

There are no failure logs or errors that I can find so I'm not sure where else to look. If my data types are mismatched from the connector converts and my producer could that cause this type of issue? That's the only thing I can see that may not quite line up, but I feel like the the connector ready status should be true like the KafkaConnect resource.

kafkaConnect yaml:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: kafka-connect-tls
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  image: <ecr url>/strimzi-snowflake-connector:152f837e
  version: 3.7.0
  replicas: 1
  authentication:
    type: tls
    certificateAndKey:
      certificate: tls.crt
      key: tls.key
      secretName: user-cert-and-key
  bootstrapServers: kafka-cluster-kafka-bootstrap:9093
  tls:
    trustedCertificates:
      - secretName: kafka-cluster-cluster-ca-cert
        certificate: ca.crt
  config:
    group.id: connect-cluster
    config.providers: env
    config.providers.env.class: org.apache.kafka.common.config.provider.EnvVarConfigProvider
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-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
    # -1 means it will use the default replication factor configured in the broker
    config.storage.replication.factor: -1
    offset.storage.replication.factor: -1
    status.storage.replication.factor: -1
  template:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
              - matchExpressions:
                - key: agentpool
                  operator: In
                  values:
                  - strimzi
  externalConfiguration:
    env:
      - name: SNOWFLAKE_URL_NAME
        valueFrom:
          configMapKeyRef:
            name: snowflake-connector-configs
            key: url_name
      - name: SNOWFLAKE_DATABASE_NAME
        valueFrom:
          configMapKeyRef:
            name: snowflake-connector-configs
            key: database_name
      - name: SNOWFLAKE_SCHEMA_NAME
        valueFrom:
          configMapKeyRef:
            name: snowflake-connector-configs
            key: schema_name
      - name: SNOWFLAKE_ROLE_NAME
        valueFrom:
          configMapKeyRef:
            name: snowflake-connector-configs
            key: role
      - name: SNOWFLAKE_USER_NAME
        valueFrom:
          secretKeyRef:
            name: snowflake-user-name
            key: user-name
      - name: SNOWFLAKE_PRIVATE_KEY
        valueFrom:
          secretKeyRef:
            name: snowflake-private-key
            key: private-key

snowflake connector yaml:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: snowflake-connector
spec:
  class: com.snowflake.kafka.connector.SnowflakeSinkConnector
  tasksMax: 8
  autoRestart:
    enabled: true
  config:
    name: snowflake-connector
    topics: test-topic
    snowflake.topic2table.map: test-topic:streaming_data
    buffer.count.records: "10000"
    buffer.flush.time: "60"
    buffer.size.bytes: "5000000"

    # configMaps
    snowflake.url.name: ${env:SNOWFLAKE_URL_NAME}
    snowflake.database.name: ${env:SNOWFLAKE_DATABASE_NAME}
    snowflake.schema.name: ${env:SNOWFLAKE_SCHEMA_NAME}
    snowflake.role.name: ${env:SNOWFLAKE_ROLE_NAME}

    # secrets
    snowflake.user.name: ${env:SNOWFLAKE_USER_NAME}
    snowflake.private.key: ${env:SNOWFLAKE_PRIVATE_KEY}

    key.converter: org.apache.kafka.connect.storage.StringConverter
    value.converter: com.snowflake.kafka.connector.records.SnowflakeJsonConverter

produce.py

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError

# internal
BOOTSTRAP_SERVER_TLS = "kafka-cluster-kafka-bootstrap.strimzi.svc:9093"

def produce():
    print('producing...')
    producer = KafkaProducer(bootstrap_servers=[BOOTSTRAP_SERVER_TLS],
                             security_protocol="SSL",
                             ssl_cafile="/mnt/secrets-store/strimzi-poc-cluster-cert.crt",
                             ssl_certfile="/mnt/secrets-store/strimzi-poc-user-crt.crt",
                             ssl_keyfile="/mnt/secrets-store/strimzi-poc-user-crt.key")

    print('producer created')
    producer.send('test-topic', key=b'foo', value=b'bar')
    producer.flush()

    print('sent message')

    # Asynchronous by default
    future = producer.send('test-topic', b'raw_bytes')
    print(future)

    # Block for 'synchronous' sends
    record_metadata = future.get(timeout=10)

    # Successful result returns assigned partition and offset
    print (record_metadata.topic)
    print (record_metadata.partition)
    print (record_metadata.offset)


produce()
# consume()

output of kubectl get kafkaconnector -o yaml:

apiVersion: v1
items:
- apiVersion: kafka.strimzi.io/v1beta2
  kind: KafkaConnector
  metadata:
    annotations:
      kubectl.kubernetes.io/last-applied-configuration: |
        {"apiVersion":"kafka.strimzi.io/v1beta2","kind":"KafkaConnector","metadata":{"annotations":{},"name":"snowflake-connector","namespace":"strimzi"},"spec":{"autoRestart":{"enabled":true},"class":"com.snowflake.kafka.connector.SnowflakeSinkConnector","config":{"buffer.count.records":"10000","buffer.flush.time":"60","buffer.size.bytes":"5000000","key.converter":"org.apache.kafka.connect.storage.StringConverter","name":"snowflake-connector","snowflake.database.name":"${env:SNOWFLAKE_DATABASE_NAME}","snowflake.private.key":"${env:SNOWFLAKE_PRIVATE_KEY}","snowflake.role.name":"${env:SNOWFLAKE_ROLE_NAME}","snowflake.schema.name":"${env:SNOWFLAKE_SCHEMA_NAME}","snowflake.topic2table.map":"test-topic:streaming_data","snowflake.url.name":"${env:SNOWFLAKE_URL_NAME}","snowflake.user.name":"${env:SNOWFLAKE_USER_NAME}","topics":"test-topic","value.converter":"com.snowflake.kafka.connector.records.SnowflakeJsonConverter"},"tasksMax":8}}
    creationTimestamp: "2024-04-22T18:09:32Z"
    generation: 1
    name: snowflake-connector
    namespace: strimzi
    resourceVersion: "38381029"
    uid: 975b9e68-ea24-4cb0-ae69-c36f49f4bfc3
  spec:
    autoRestart:
      enabled: true
    class: com.snowflake.kafka.connector.SnowflakeSinkConnector
    config:
      buffer.count.records: "10000"
      buffer.flush.time: "60"
      buffer.size.bytes: "5000000"
      key.converter: org.apache.kafka.connect.storage.StringConverter
      name: snowflake-connector
      snowflake.database.name: ${env:SNOWFLAKE_DATABASE_NAME}
      snowflake.private.key: ${env:SNOWFLAKE_PRIVATE_KEY}
      snowflake.role.name: ${env:SNOWFLAKE_ROLE_NAME}
      snowflake.schema.name: ${env:SNOWFLAKE_SCHEMA_NAME}
      snowflake.topic2table.map: test-topic:streaming_data
      snowflake.url.name: ${env:SNOWFLAKE_URL_NAME}
      snowflake.user.name: ${env:SNOWFLAKE_USER_NAME}
      topics: test-topic
      value.converter: com.snowflake.kafka.connector.records.SnowflakeJsonConverter
    tasksMax: 8
kind: List
metadata:
  resourceVersion: ""

output of kubectl get kafkaconnect -o yaml:

apiVersion: v1
items:
- apiVersion: kafka.strimzi.io/v1beta2
  kind: KafkaConnect
  metadata:
    annotations:
      kubectl.kubernetes.io/last-applied-configuration: |
        {"apiVersion":"kafka.strimzi.io/v1beta2","kind":"KafkaConnect","metadata":{"annotations":{"strimzi.io/use-connector-resources":"true"},"name":"kafka-connect-tls","namespace":"strimzi"},"spec":{"authentication":{"certificateAndKey":{"certificate":"tls.crt","key":"tls.key","secretName":"user-cert-and-key"},"type":"tls"},"bootstrapServers":"kafka-cluster-kafka-bootstrap:9093","config":{"config.providers":"env","config.providers.env.class":"org.apache.kafka.common.config.provider.EnvVarConfigProvider","config.storage.replication.factor":-1,"config.storage.topic":"connect-cluster-configs","group.id":"connect-cluster","key.converter":"org.apache.kafka.connect.storage.StringConverter","key.converter.schemas.enable":false,"offset.storage.replication.factor":-1,"offset.storage.topic":"connect-cluster-offsets","status.storage.replication.factor":-1,"status.storage.topic":"connect-cluster-status","value.converter":"org.apache.kafka.connect.json.JsonConverter","value.converter.schemas.enable":false},"externalConfiguration":{"env":[{"name":"SNOWFLAKE_URL_NAME","valueFrom":{"configMapKeyRef":{"key":"url_name","name":"snowflake-connector-configs"}}},{"name":"SNOWFLAKE_DATABASE_NAME","valueFrom":{"configMapKeyRef":{"key":"database_name","name":"snowflake-connector-configs"}}},{"name":"SNOWFLAKE_SCHEMA_NAME","valueFrom":{"configMapKeyRef":{"key":"schema_name","name":"snowflake-connector-configs"}}},{"name":"SNOWFLAKE_ROLE_NAME","valueFrom":{"configMapKeyRef":{"key":"role","name":"snowflake-connector-configs"}}},{"name":"SNOWFLAKE_USER_NAME","valueFrom":{"secretKeyRef":{"key":"user-name","name":"snowflake-user-name"}}},{"name":"SNOWFLAKE_PRIVATE_KEY","valueFrom":{"secretKeyRef":{"key":"private-key","name":"snowflake-private-key"}}}]},"image":"<acr url>/strimzi-snowflake-connector:152f837e","replicas":1,"template":{"pod":{"affinity":{"nodeAffinity":{"requiredDuringSchedulingIgnoredDuringExecution":{"nodeSelectorTerms":[{"matchExpressions":[{"key":"agentpool","operator":"In","values":["strimzi"]}]}]}}}}},"tls":{"trustedCertificates":[{"certificate":"ca.crt","secretName":"kafka-cluster-cluster-ca-cert"}]},"version":"3.7.0"}}
      strimzi.io/use-connector-resources: "true"
    creationTimestamp: "2024-04-22T15:24:03Z"
    generation: 4
    name: kafka-connect-tls
    namespace: strimzi
    resourceVersion: "38372944"
    uid: 7a77ee95-17a8-4150-81c5-dfcade6d4f2c
  spec:
    authentication:
      certificateAndKey:
        certificate: tls.crt
        key: tls.key
        secretName: user-cert-and-key
      type: tls
    bootstrapServers: kafka-cluster-kafka-bootstrap:9093
    config:
      config.providers: env
      config.providers.env.class: org.apache.kafka.common.config.provider.EnvVarConfigProvider
      config.storage.replication.factor: -1
      config.storage.topic: connect-cluster-configs
      group.id: connect-cluster
      key.converter: org.apache.kafka.connect.storage.StringConverter
      key.converter.schemas.enable: false
      offset.storage.replication.factor: -1
      offset.storage.topic: connect-cluster-offsets
      status.storage.replication.factor: -1
      status.storage.topic: connect-cluster-status
      value.converter: org.apache.kafka.connect.json.JsonConverter
      value.converter.schemas.enable: false
    externalConfiguration:
      env:
      - name: SNOWFLAKE_URL_NAME
        valueFrom:
          configMapKeyRef:
            key: url_name
            name: snowflake-connector-configs
      - name: SNOWFLAKE_DATABASE_NAME
        valueFrom:
          configMapKeyRef:
            key: database_name
            name: snowflake-connector-configs
      - name: SNOWFLAKE_SCHEMA_NAME
        valueFrom:
          configMapKeyRef:
            key: schema_name
            name: snowflake-connector-configs
      - name: SNOWFLAKE_ROLE_NAME
        valueFrom:
          configMapKeyRef:
            key: role
            name: snowflake-connector-configs
      - name: SNOWFLAKE_USER_NAME
        valueFrom:
          secretKeyRef:
            key: user-name
            name: snowflake-user-name
      - name: SNOWFLAKE_PRIVATE_KEY
        valueFrom:
          secretKeyRef:
            key: private-key
            name: snowflake-private-key
    image: <acr url>/strimzi-snowflake-connector:152f837e
    replicas: 1
    template:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
              - matchExpressions:
                - key: agentpool
                  operator: In
                  values:
                  - strimzi
    tls:
      trustedCertificates:
      - certificate: ca.crt
        secretName: kafka-cluster-cluster-ca-cert
    version: 3.7.0
  status:
    conditions:
    - lastTransitionTime: "2024-04-22T17:41:32.036161517Z"
      status: "True"
      type: Ready
    connectorPlugins:
    - class: com.snowflake.kafka.connector.SnowflakeSinkConnector
      type: sink
      version: 2.2.1
    - class: org.apache.kafka.connect.mirror.MirrorCheckpointConnector
      type: source
      version: 3.7.0
    - class: org.apache.kafka.connect.mirror.MirrorHeartbeatConnector
      type: source
      version: 3.7.0
    - class: org.apache.kafka.connect.mirror.MirrorSourceConnector
      type: source
      version: 3.7.0
    labelSelector: strimzi.io/cluster=kafka-connect-tls,strimzi.io/name=kafka-connect-tls-connect,strimzi.io/kind=KafkaConnect
    observedGeneration: 4
    replicas: 1
    url: http://kafka-connect-tls-connect-api.strimzi.svc:8083
kind: List
metadata:
  resourceVersion: ""

Dockerfile:

FROM quay.io/strimzi/kafka:0.40.0-kafka-3.7.0
USER root:root
COPY ./snowflake-kafka-connector-2.2.1.jar /opt/kafka/plugins/snowflake-kafka-sink-connector/snowflake-kafka-connector-2.2.1.jar
USER 1001

[scholzj]

I guess you are missing the strimzi.io/cluster label?

[rick-reihl]

The kafkaConnect pod is running and has the labels listed in the labelSelector section above.

kubectl describe po kafka-connect-tls-connnect-0 yeilds

Name:             kafka-connect-tls-connect-0
Namespace:        strimzi
Priority:         0
Service Account:  kafka-connect-tls-connect
Node:             aks-strimzi-16549309-vmss000003/172.20.48.243
Start Time:       Mon, 22 Apr 2024 17:40:20 +0000
Labels:           app.kubernetes.io/instance=kafka-connect-tls
                  app.kubernetes.io/managed-by=strimzi-cluster-operator
                  app.kubernetes.io/name=kafka-connect
                  app.kubernetes.io/part-of=strimzi-kafka-connect-tls
                  statefulset.kubernetes.io/pod-name=kafka-connect-tls-connect-0
                  strimzi.io/cluster=kafka-connect-tls
                  strimzi.io/component-type=kafka-connect
                  strimzi.io/controller=strimzipodset
                  strimzi.io/controller-name=kafka-connect-tls-connect
                  strimzi.io/kind=KafkaConnect
                  strimzi.io/name=kafka-connect-tls-connect
                  strimzi.io/pod-name=kafka-connect-tls-connect-0
Annotations:      strimzi.io/auth-hash: -1476375049
                  strimzi.io/logging-appenders-hash: 06ee78c4
                  strimzi.io/revision: 488efaf2
Status:           Running
IP:               172.20.48.39
IPs:
  IP:           172.20.48.39
Controlled By:  StrimziPodSet/kafka-connect-tls-connect
Containers:
  kafka-connect-tls-connect:
    Container ID:  containerd://71019af2dea4ffd988d2e9b3997807cad76ec5e59fe59d8f6ce72d6e9ea0c03c
    Image:         <acr>/strimzi-snowflake-connector:152f837e
    Image ID:      <acr>/strimzi-snowflake-connector@sha256:1867258f7a81087ab6cb38d587366f113ff6fa062a1cb30e92d896e504716d9d
    Port:          8083/TCP
    Host Port:     0/TCP
    Args:
      /opt/kafka/kafka_connect_run.sh
    State:          Running
      Started:      Mon, 22 Apr 2024 17:40:24 +0000
    Ready:          True
    Restart Count:  0
    Liveness:       http-get http://:rest-api/ delay=60s timeout=5s period=10s #success=1 #failure=3
    Readiness:      http-get http://:rest-api/ delay=60s timeout=5s period=10s #success=1 #failure=3
    Environment:
      KAFKA_CONNECT_CONFIGURATION:      offset.storage.topic=connect-cluster-offsets
                                        value.converter=org.apache.kafka.connect.json.JsonConverter
                                        config.storage.topic=connect-cluster-configs
                                        key.converter=org.apache.kafka.connect.storage.StringConverter
                                        group.id=connect-cluster
                                        status.storage.topic=connect-cluster-status
                                        config.providers=env
                                        config.providers.env.class=org.apache.kafka.common.config.provider.EnvVarConfigProvider
                                        config.storage.replication.factor=-1
                                        key.converter.schemas.enable=false
                                        offset.storage.replication.factor=-1
                                        status.storage.replication.factor=-1
                                        value.converter.schemas.enable=false

      KAFKA_CONNECT_METRICS_ENABLED:    false
      KAFKA_CONNECT_BOOTSTRAP_SERVERS:  kafka-cluster-kafka-bootstrap:9093
      STRIMZI_KAFKA_GC_LOG_ENABLED:     false
      KAFKA_HEAP_OPTS:                  -Xms128M
      KAFKA_CONNECT_TLS:                true
      KAFKA_CONNECT_TRUSTED_CERTS:      kafka-cluster-cluster-ca-cert/ca.crt
      KAFKA_CONNECT_TLS_AUTH_CERT:      user-cert-and-key/tls.crt
      KAFKA_CONNECT_TLS_AUTH_KEY:       user-cert-and-key/tls.key
      SNOWFLAKE_URL_NAME:               <set to the key 'url_name' of config map 'snowflake-connector-configs'>       Optional: false
      SNOWFLAKE_DATABASE_NAME:          <set to the key 'database_name' of config map 'snowflake-connector-configs'>  Optional: false
      SNOWFLAKE_SCHEMA_NAME:            <set to the key 'schema_name' of config map 'snowflake-connector-configs'>    Optional: false
      SNOWFLAKE_ROLE_NAME:              <set to the key 'role' of config map 'snowflake-connector-configs'>           Optional: false
      SNOWFLAKE_USER_NAME:              <set to the key 'user-name' in secret 'snowflake-user-name'>          Optional: false
      SNOWFLAKE_PRIVATE_KEY:            <set to the key 'private-key' in secret 'snowflake-private-key'>      Optional: false
    Mounts:
      /opt/kafka/connect-certs/user-cert-and-key from user-cert-and-key (rw)
      /opt/kafka/connect-certs/kafka-cluster-cluster-ca-cert from kafka-cluster-cluster-ca-cert (rw)
      /opt/kafka/custom-config/ from kafka-metrics-and-logging (rw)
      /tmp from strimzi-tmp (rw)
      /var/run/secrets/kubernetes.io/serviceaccount from kube-api-access-667db (ro)
Conditions:
  Type              Status
  Initialized       True
  Ready             True
  ContainersReady   True
  PodScheduled      True
Volumes:
  strimzi-tmp:
    Type:       EmptyDir (a temporary directory that shares a pod's lifetime)
    Medium:     Memory
    SizeLimit:  5Mi
  kafka-metrics-and-logging:
    Type:      ConfigMap (a volume populated by a ConfigMap)
    Name:      kafka-connect-tls-connect-config
    Optional:  false
  kafka-cluster-cluster-ca-cert:
    Type:        Secret (a volume populated by a Secret)
    SecretName:  kafka-cluster-cluster-ca-cert
    Optional:    false
  user-cert-and-key:
    Type:        Secret (a volume populated by a Secret)
    SecretName:  user-cert-and-key
    Optional:    false
  kube-api-access-667db:
    Type:                    Projected (a volume that contains injected data from multiple sources)
    TokenExpirationSeconds:  3607
    ConfigMapName:           kube-root-ca.crt
    ConfigMapOptional:       <nil>
    DownwardAPI:             true
QoS Class:                   BestEffort
Node-Selectors:              <none>
Tolerations:                 node.kubernetes.io/not-ready:NoExecute op=Exists for 300s
                             node.kubernetes.io/unreachable:NoExecute op=Exists for 300s
Events:                      <none>

I did notice the kafka-cluster-entity-operator is on a different node than I intended, but I don't expect that to have any impact on strimzi functionality.
kubect get po -o wide

NAME                                             READY   STATUS    RESTARTS       AGE   IP              NODE                              NOMINATED NODE   READINESS GATES
kafka-cluster-entity-operator-5b7bd986f6-g7bnf   2/2     Running   11 (13d ago)   13d   172.20.48.19    aks-user2-28074808-vmss00000j     <none>           <none>
kafka-cluster-kafka-0                            1/1     Running   0              23h   172.20.48.244   aks-strimzi-16549309-vmss000003   <none>           <none>
kafka-cluster-zookeeper-0                        1/1     Running   0              13d   172.20.48.251   aks-strimzi-16549309-vmss000003   <none>           <none>
kafka-connect-tls-connect-0                      1/1     Running   0              21h   172.20.48.39    aks-strimzi-16549309-vmss000003   <none>           <none>
strimzi-cluster-operator-f57f4758f-sh58t         1/1     Running   1 (25h ago)    13d   172.20.48.57    aks-strimzi-16549309-vmss000004   <none>           <none>
strimzi-test-55bf65b555-6dcxn                    1/1     Running   0              23h   172.20.48.33    aks-strimzi-16549309-vmss000003   <none>           <none>

Am I correct in assuming that the KafkaConnector should say READY true? I'm stumped.

[scholzj]

I guess you are missing the strimzi.io/cluster label?

Sorry if this was not clear. But you need to have this label on the KafkaConnector resource and it has to point to the name of the KafkaConnect resource it should be used with.

[rick-reihl]

Thank you SO much for your prompt help. That seems to have been the missing link.