kafkaconnector: Error initializing the ssl context for RestService

[varma519]

Unable to bring kafkaconnector with secure schema registry.

Config trail-1

Followed steps at https://github.com/strimzi/strimzi-kafka-operator/blob/main/documentation/api/io.strimzi.api.kafka.model.connect.ExternalConfiguration.adoc#volumes.

Kafkaconnect pod has below error:

 ERROR [rawstream-sf-sink-connector|task-0] Modification time of key store could not be obtained: ${directory:/opt/kafka/external-configuration/kafka-connector/kafka-connector:keystore.jks} (io.confluent.kafka.schemaregistry.client.security.SslFactory) [StartAndStopExecutor-connect-1-2]
java.nio.file.NoSuchFileException: ${directory:/opt/kafka/external-configuration/kafka-connector/kafka-connector:keystore.jks}
	at java.base/sun.nio.fs.UnixException.translateToIOException(UnixException.java:92)
	at java.base/sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:106)
	at java.base/sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:111)
	at java.base/sun.nio.fs.UnixFileAttributeViews$Basic.readAttributes(UnixFileAttributeViews.java:55)
	at java.base/sun.nio.fs.UnixFileSystemProvider.readAttributes(UnixFileSystemProvider.java:148)
	at java.base/sun.nio.fs.LinuxFileSystemProvider.readAttributes(LinuxFileSystemProvider.java:99)
	at java.base/java.nio.file.Files.readAttributes(Files.java:1851)
	at java.base/java.nio.file.Files.getLastModifiedTime(Files.java:2402)

Here is the snapshot of config:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
  namespace: my-cluster-operator-namespace
spec:
  version: 3.3.1
  replicas: 1
  authentication:
    type: tls
    certificateAndKey:
      certificate: cert.pem
      key: key.pem
      secretName: kafka-connect-secrets-2
  bootstrapServers: <server details>
  tls:
    trustedCertificates:
      - secretName: kafka-connect-secrets-2
        certificate: manual_open_server
  config:
    group.id: <gid>
    offset.storage.topic: <topicname>
    config.storage.topic: <topicname>
    status.storage.topic: <topicname>
    config.storage.replication.factor: 3
    offset.storage.replication.factor: 3
    status.storage.replication.factor: 3
    config.providers: directory
    config.providers.file.class: org.apache.kafka.common.config.provider.DirectoryConfigProvider
  externalConfiguration:
    volumes:
      - name: kafka-connector
        secret:
          secretName: kafka-connector

Here is connector config,

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  labels:
    strimzi.io/cluster: my-connect-cluster
  name: rawstream-sf-sink-connector
  namespace: my-cluster-operator-namespace
spec:
  class: com.snowflake.kafka.connector.SnowflakeSinkConnector
  tasksMax: 1
  config:
    consumer.override.bootstrap.servers: <server:ip>
    consumer.override.group.id: <gid>
    topics: <topic>
    consumer.override.security.protocol: SSL
    consumer.override.enable.ssl.certificate.verification : false
    consumer.override.sasl.mechanism: PLAIN
    consumer.override.ssl.truststore.location: "${directory:/opt/kafka/external-configuration/kafka-connector/kafka-connector:truststore.jks}"
    consumer.override.ssl.truststore.password: <pd>
    consumer.override.ssl.keystore.location: "${directory:/opt/kafka/external-configuration/kafka-connector/kafka-connector:keystore.jks}"
    consumer.override.ssl.keystore.password: <pd>
    consumer.override.request.timeout.ms: 80000
    consumer.override.schema.registry.url: <srurl>
    key.converter: com.snowflake.kafka.connector.records.SnowflakeAvroConverter
    value.converter: com.snowflake.kafka.connector.records.SnowflakeAvroConverter
    key.converter.schemas.enable: false
    value.converter.schemas.enable: false
    key.converter.schema.registry.url: <srurl>
    key.converter.schema.registry.ssl.truststore.location: "${directory:/opt/kafka/external-configuration/kafka-connector/kafka-connector:truststore.jks}"
    key.converter.schema.registry.ssl.truststore.password: <pd>
    key.converter.schema.registry.ssl.keystore.location: "${directory:/opt/kafka/external-configuration/kafka-connector/kafka-connector:keystore.jks}"
    key.converter.schema.registry.ssl.keystore.password: <pd>
    key.converter.schema.registry.ssl.key.password: <pd>
    value.converter.schema.registry.url: <srurl>
    value.converter.schema.registry.ssl.truststore.location: "${directory:/opt/kafka/external-configuration/kafka-connector/kafka-connector:truststore.jks}"
    value.converter.schema.registry.ssl.truststore.password: <pd>
    value.converter.schema.registry.ssl.keystore.location: "${directory:/opt/kafka/external-configuration/kafka-connector/kafka-connector:keystore.jks}"
    value.converter.schema.registry.ssl.keystore.password: <pd>
    value.converter.schema.registry.ssl.key.password: <pd>
    buffer.flush.time: 60
    buffer.count.records: 500
    buffer.size.bytes: 1000000
    snowflake.url.name: <url>
    snowflake.user.name: <name>
    snowflake.role.name: <role>
    snowflake.private.key: <key>
    snowflake.database.name: DEMO_DB
    snowflake.schema.name: PUBLIC
    snowflake.ingestion.method: SNOWPIPE
    snowflake.topic2table.map: <topic:table>
    jmx: true

I changed config like below to address 'NoSuchFileException'

Config trail-2

Pod log still has schema issue, but path issue is not sure and avroconverterconfig looks fine.

2023-03-03 22:27:47,802 INFO [rawstream-sf-sink-connector|task-0] AvroConverterConfig values:
	auto.register.schemas = true
	basic.auth.credentials.source = URL
	basic.auth.user.info = [hidden]
	bearer.auth.credentials.source = STATIC_TOKEN
	bearer.auth.token = [hidden]
	context.name.strategy = class io.confluent.kafka.serializers.context.NullContextNameStrategy
	id.compatibility.strict = true
	key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
	latest.compatibility.strict = true
	max.schemas.per.subject = 1000
	normalize.schemas = false
	proxy.host =
	proxy.port = -1
	schema.reflection = false
	schema.registry.basic.auth.user.info = [hidden]
	schema.registry.ssl.cipher.suites = null
	schema.registry.ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
	schema.registry.ssl.endpoint.identification.algorithm = https
	schema.registry.ssl.engine.factory.class = null
	schema.registry.ssl.key.password = [hidden]
	schema.registry.ssl.keymanager.algorithm = SunX509
	schema.registry.ssl.keystore.certificate.chain = null
	schema.registry.ssl.keystore.key = null
	schema.registry.ssl.keystore.location = /opt/kafka/external-configuration/kafka-connector/keystore.jks
	schema.registry.ssl.keystore.password = [hidden]
	schema.registry.ssl.keystore.type = JKS
	schema.registry.ssl.protocol = TLSv1.3
	schema.registry.ssl.provider = null
	schema.registry.ssl.secure.random.implementation = null
	schema.registry.ssl.trustmanager.algorithm = PKIX
	schema.registry.ssl.truststore.certificates = null
	schema.registry.ssl.truststore.location = /opt/kafka/external-configuration/kafka-connector/truststore.jks
	schema.registry.ssl.truststore.password = [hidden]
	schema.registry.ssl.truststore.type = JKS
	schema.registry.url = : https://<secure schema registery.. removed intentionally >.com
	use.latest.version = false
	use.schema.id = -1
	value.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
 (com.snowflake.kafka.connector.records.AvroConverterConfig) [StartAndStopExecutor-connect-1-6]
2023-03-03 22:27:47,803 ERROR [rawstream-sf-sink-connector|task-0] Failed to start task rawstream-sf-sink-connector-0 (org.apache.kafka.connect.runtime.Worker) [StartAndStopExecutor-connect-1-6]
com.snowflake.kafka.connector.internal.SnowflakeKafkaConnectorException: [SF_KAFKA_CONNECTOR] Exception: Failed to connect schema registry service
Error Code: 0012
Detail: Schema registry service is not available
Message: Error initializing the ssl context for RestService
io.confluent.kafka.schemaregistry.client.security.SslFactory.<init>(SslFactory.java:117)
io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.<init>(CachedSchemaRegistryClient.java:231)
io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.<init>(CachedSchemaRegistryClient.java:174)
io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.<init>(CachedSchemaRegistryClient.java:148)
io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.<init>(CachedSchemaRegistryClient.java:109)
com.snowflake.kafka.connector.records.SnowflakeAvroConverter.configure(SnowflakeAvroConverter.java:60)
org.apache.kafka.connect.runtime.isolation.Plugins.newConverter(Plugins.java:297)

Our Schema registry is secure. I am able to reach to Schema registry with curl from kafka-connect pod. So, no connection issue.
jks files are content encoded with base64 for keystore and truststore.

Does connector really able to read the certs? does it still has certs config issue or am i missing any other config ?

[varma519]

@scholzj/anyone,
do you have any guess what went wrong here ? any help much appreciated.

[kyguy]

Hi @varma519,

It looks like the keystore and truststore paths in the KafkaConnector resource are misconfigured. Double check the content of the "kafka-connector" Secret to make sure you have the right truststore and keystore file names and types.

In the docs example, PEM format is used [1], it might be easier to stick to that format and example to get it working first before trying JKS format (which it appears to be the format your KafkaConnector is being configured to use from the log provided above)

Follow up with me on the #strimzi Slack channel [2], it'll be easier to discuss the issue there

[1] https://github.com/strimzi/strimzi-kafka-operator/blob/main/documentation/api/io.strimzi.api.kafka.model.connect.ExternalConfiguration.adoc#volumes
[2] https://communityinviter.com/apps/cloud-native/cncf

[varma519]

Thank you so much @kyguy. Let me try PEM and will touch base on slack channel further.

[scholzj]

The exception suggests NoSuchFileException -> so that would normally suggest that the path is wrong or the file does not exist. Did you tried to exec into the Connect Pod and check if the file is there on given path? You should also consult the Schema Registry docs with regards to what it expects. But the DirectoryConfigProvider will give you content of the file. However in Kafka, the .location configuration options usually expect to get a path. So I think even if the file existed, it would not work as it might not expect you to just dump the JSK content into the field.

[andreyolv]

Hello @varma519
How did you save your .jks files as secrets?
I tried but it gives an error saying that secret cannot save because the file contains null values, so I save them in a configmap.
Thanks