how to override kafkaconnect's acl configuration in kafkaconnector

[tusharbhasme]

We need to use a different user/secret per connector in a multi-tenant kafka-connect setup. I have added the KubernetesSecretConfigProvider in my kafka-connect:

config.providers=secrets
config.providers.secrets.class=io.strimzi.kafka.KubernetesSecretConfigProvider

I am trying below configuration to override the keys in connector:

consumer.override.ssl.keystore.location: ${secrets:kafka/user-splunk-connector-0a28f01e-4468-4b11-8064-7ede39a80047:user.p12}
consumer.override.ssl.keystore.password: ${secrets:kafka/user-splunk-connector-0a28f01e-4468-4b11-8064-7ede39a80047:user.password}

but I am getting below exception in the kafka-connect the moment the connector is deployed:

│       State:  FAILED                                                                                                                                                                                                                       │
│       Trace:  java.lang.NullPointerException                                                                                                                                                                                               │
│               at org.apache.kafka.connect.runtime.Worker$ConnectorStatusMetricsGroup.recordTaskRemoved(Worker.java:1043)                                                                                                                   │
│               at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:558)                                                                                                                                                        │
│               at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startTask(DistributedHerder.java:1421)                                                                                                                     │
│               at org.apache.kafka.connect.runtime.distributed.DistributedHerder.lambda$getTaskStartingCallable$22(DistributedHerder.java:1434)                                                                                             │
│               at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)                                                                                                                                                        │
│               at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)                                                                                                                                 │
│               at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)                                                                                                                                 │
│               at java.base/java.lang.Thread.run(Thread.java:829)                                                                                                                                                                           │
│                                                                                                                                                                                                                                            │
│       worker_id:  10.2.27.110:8083                                                                                                                                                                                                         │
│       Id:         1                                                                                                                                                                                                                        │
│       State:      FAILED                                                                                                                                                                                                                   │
│       Trace:      org.apache.kafka.common.KafkaException: Failed to construct kafka consumer                                                                                                                                               │
│                   at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:823)                                                                                                                                        │
│                   at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:664)                                                                                                                                        │
│                   at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:612)                                                                                                                                        │
│                   at org.apache.kafka.connect.runtime.Worker.buildWorkerTask(Worker.java:633)                                                                                                                                              │
│                   at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:549)                                                                                                                                                    │
│                   at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startTask(DistributedHerder.java:1421)                                                                                                                 │
│                   at org.apache.kafka.connect.runtime.distributed.DistributedHerder.lambda$getTaskStartingCallable$22(DistributedHerder.java:1434)                                                                                         │
│                   at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)                                                                                                                                                    │
│                   at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)                                                                                                                             │
│                   at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)                                                                                                                             │
│                   at java.base/java.lang.Thread.run(Thread.java:829)                                                                                                                                                                       │
│ Caused by: org.apache.kafka.common.KafkaException: java.nio.file.InvalidPathException: Nul character not allowed: 0� 
 <.... lots of non-selectable binary data>
│ ���mu�Ӫ�����;o� �ݰPo0�d���K\!h�����qn�Q����C�7�dL����52��Ӑ�1��0#  *�H��  1���g�f��Yr�hF��<�0��  *�H��  1vtuser-splunk-connector-0a28f01e-4468-4b11-8064-7ede39a80047010!0  +I&��P�J�k,�v��  *8��4[8�&                                      │
│                                                                         at java.base/sun.nio.fs.UnixPath.checkNotNul(UnixPath.java:91)                                                                                                     │
│                                                                         at java.base/sun.nio.fs.UnixPath.normalizeAndCheck(UnixPath.java:81)                                                                                               │
│                                                                         at java.base/sun.nio.fs.UnixPath.<init>(UnixPath.java:69)                                                                                                          │
│                                                                         at java.base/sun.nio.fs.UnixFileSystem.getPath(UnixFileSystem.java:279)                                                                                            │
│                                                                         at java.base/java.nio.file.Path.of(Path.java:147)                                                                                                                  │
│                                                                         at java.base/java.nio.file.Paths.get(Paths.java:69)                                                                                                                │
│                                                                         at org.apache.kafka.common.security.ssl.DefaultSslEngineFactory$FileBasedStore.lastModifiedMs(DefaultSslEngineFactory.java:383)                                    │
│                                                                         at org.apache.kafka.common.security.ssl.DefaultSslEngineFactory$FileBasedStore.<init>(DefaultSslEngineFactory.java:348)                                            │
│                                                                         at org.apache.kafka.common.security.ssl.DefaultSslEngineFactory.createKeystore(DefaultSslEngineFactory.java:299)                                                   │
│                                                                         at org.apache.kafka.common.security.ssl.DefaultSslEngineFactory.configure(DefaultSslEngineFactory.java:161)                                                        │
│                                                                         at org.apache.kafka.common.security.ssl.SslFactory.instantiateSslEngineFactory(SslFactory.java:140)                                                                │
│                                                                         at org.apache.kafka.common.security.ssl.SslFactory.configure(SslFactory.java:97)                                                                                   │
│                                                                         at org.apache.kafka.common.network.SslChannelBuilder.configure(SslChannelBuilder.java:73)                                                                          │
│                                                                         ... 14 more                            

Is it not able to load a pkcs12 file?

If I try to directly use the pem files:

consumer.override.ssl.keystore.type: PEM
consumer.override.ssl.keystore.certificate.chain: ${secrets:kafka/user-splunk-connector-0a28f01e-4468-4b11-8064-7ede39a80047:user.crt}
consumer.override.ssl.keystore.key: ${secrets:kafka/user-splunk-connector-0a28f01e-4468-4b11-8064-7ede39a80047:user.key}

I get stuck with a different error:

│       Trace:      org.apache.kafka.common.KafkaException: Failed to construct kafka consumer                                                                                                                                               │
│                   at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:823)                                                                                                                                        │
│                   at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:664)                                                                                                                                        │
│                   at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:612)                                                                                                                                        │
│                   at org.apache.kafka.connect.runtime.Worker.buildWorkerTask(Worker.java:633)                                                                                                                                              │
│                   at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:549)                                                                                                                                                    │
│                   at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startTask(DistributedHerder.java:1421)                                                                                                                 │
│                   at org.apache.kafka.connect.runtime.distributed.DistributedHerder.lambda$getTaskStartingCallable$22(DistributedHerder.java:1434)                                                                                         │
│                   at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)                                                                                                                                                    │
│                   at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)                                                                                                                             │
│                   at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)                                                                                                                             │
│                   at java.base/java.lang.Thread.run(Thread.java:829)                                                                                                                                                                       │
│ Caused by: org.apache.kafka.common.errors.InvalidConfigurationException: Both SSL key store location and separate private key are specified.

Seems keystore.location gets set at somewhere in kafakconnect configuration and cannot be overridden here in kafkaconnector.

[scholzj]

I think there is some previous discussion or issue about this already. When the Kafka option requires location to a certificate store, you cannot use the secret provider. You have to mount the secret as volume into the Connect pod and specify the path to the keystore / truststore.

[tusharbhasme]

Thanks @scholzj, yes you are right. I had raised an issue around this topic #5189 which was closed with a suggestion. I followed that suggestion and had follow up questions, thus started this discussion thread.

Thing is, the kafkaconnectors are to be deployed dynamically for each tenant being onboarded and dynamically patching kafkaconnect with the secret mounts proved to be very problematic. On every patch, it gets into a rebalancing mode and takes a lot of time to get back to normal. At some point, it just started to fail, even with multiple kafkaconnect nodes in the cluster.

We currently are deploying a connect per tenant, which is now adding to the cost, hence we are again exploring if we can go back to multitenant connect deployment.

Coming back to the issue, since certificate store cannot be read by a secret provider, why can't we use the key directly, as I tried in the second option (using PEM format). Right now, keystore.location is being set by kafkaconnect acl which seems to be using PKCS12. So there are two options that I see:

1. We can allow keystore.location to be overridden as null in the kafkaconnector.
2. We can allow kafkaconnect to use PEM format for setting up acl, so that all properties can be properly overridden by the kafkaconnector.

[scholzj]

We currently are deploying a connect per tenant, which is now adding to the cost, hence we are again exploring if we can go back to multitenant connect deployment.

To be honest, I'm not sure what kind of multitenancy are you looking for. But I think there are way more tenancy issues then this. So if it si multitenancy you are on to, I think separate connect instances are the right way.

We can allow kafkaconnect to use PEM format for setting up acl, so that all properties can be properly overridden by the kafkaconnector.

This is something what helps you, but screws many other things. It will change the way this works for everyone who overrides the truststore / keystore. It will also break running on FIPS clusters.

We can allow keystore.location to be overridden as null in the kafkaconnector.

I do not have problem with this if you figure how to do it. I don't know how to reset the option.