search

Apicurio / apicurio-registry

An API/Schema registry - stores APIs and Schemas.
https://www.apicur.io/registry/
Apache License 2.0
609 stars 269 forks source link

Configure Kafka serdes to use a cert (trust) when communicating with Apicurio #3266

Open sgleser opened 1 year ago

sgleser commented 1 year ago

Feature or Problem Description

Add a configuration variable which points to a certificate (PEM or trust) which will be used for the communication with Apicurio Registry by the serdes.

Proposed Solution

Add a configuration beside other security configs, which are listed here in the Apicurio Registry documentation. With that setting the Kafka Connectors can have a secure communication with Apicurio Registry.

Additional Context

Adding the cert to the JVM truststore is a bad practice, because in that case other Java parts could access it and reuse it for some other stuff. Therefore it would be important that the cert is used only for the purpose of Apicurio registry communication within the Kafka Connect container.

pantaoran commented 10 months ago

I think I'm experiencing this limitation right now. I'm trying to have my Kafka Connect client (Debezium Oracle Source connector) use the Apicurio Registry client library for the schemas like this:

value.converter.apicurio.registry.converter.deserializer: io.apicurio.registry.serde.avro.AvroKafkaDeserializer
value.converter.apicurio.registry.converter.serializer: io.apicurio.registry.serde.avro.AvroKafkaSerializer
value.converter.apicurio.registry.global-id: io.apicurio.registry.utils.serde.strategy.GetOrCreateIdStrategy
value.converter.apicurio.registry.request.ssl.truststore.location: /opt/kafka/connect-certs/ca.crt
value.converter.apicurio.registry.request.ssl.truststore.type: PEM

However, that results in the following stacktrace:

java.lang.IllegalStateException: java.lang.IllegalStateException: java.security.KeyStoreException: PEM not found
    at io.apicurio.registry.resolver.AbstractSchemaResolver.configure(AbstractSchemaResolver.java:92)
    at io.apicurio.registry.resolver.DefaultSchemaResolver.configure(DefaultSchemaResolver.java:61)
    at io.apicurio.registry.serde.SchemaResolverConfigurer.configure(SchemaResolverConfigurer.java:87)
    at io.apicurio.registry.serde.AbstractKafkaSerDe.configure(AbstractKafkaSerDe.java:70)
    at io.apicurio.registry.serde.avro.AvroKafkaSerializer.configure(AvroKafkaSerializer.java:91)
    at io.apicurio.registry.utils.converter.SerdeBasedConverter.configure(SerdeBasedConverter.java:84)
    at io.apicurio.registry.utils.converter.AvroConverter.configure(AvroConverter.java:72)
    at org.apache.kafka.connect.runtime.isolation.Plugins.newConverter(Plugins.java:324)
    at org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:616)
    at org.apache.kafka.connect.runtime.Worker.startSourceTask(Worker.java:544)
    at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startTask(DistributedHerder.java:1756)
    at org.apache.kafka.connect.runtime.distributed.DistributedHerder.lambda$getTaskStartingCallable$31(DistributedHerder.java:1773)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
    at java.base/java.lang.Thread.run(Thread.java:833)\nCaused by: java.lang.IllegalStateException: java.security.KeyStoreException: PEM not found
    at io.apicurio.rest.client.JdkHttpClient.addSSL(JdkHttpClient.java:111)
    at io.apicurio.rest.client.JdkHttpClient.handleConfiguration(JdkHttpClient.java:76)
    at io.apicurio.rest.client.JdkHttpClient.<init>(JdkHttpClient.java:65)
    at io.apicurio.rest.client.JdkHttpClientProvider.create(JdkHttpClientProvider.java:14)
    at io.apicurio.rest.client.spi.ApicurioHttpClientFactory.create(ApicurioHttpClientFactory.java:26)
    at io.apicurio.registry.rest.client.RegistryClientFactory.create(RegistryClientFactory.java:79)
    at io.apicurio.registry.rest.client.RegistryClientFactory.create(RegistryClientFactory.java:67)
    at io.apicurio.registry.resolver.AbstractSchemaResolver.configure(AbstractSchemaResolver.java:88)
    ... 15 more
Caused by: java.security.KeyStoreException: PEM not found
    at java.base/java.security.KeyStore.getInstance(KeyStore.java:871)
    at io.apicurio.rest.client.JdkHttpClient.getTrustManagers(JdkHttpClient.java:120)
    at io.apicurio.rest.client.JdkHttpClient.addSSL(JdkHttpClient.java:97)
    ... 22 more
Caused by: java.security.NoSuchAlgorithmException: PEM KeyStore not available
    at java.base/sun.security.jca.GetInstance.getInstance(GetInstance.java:159)
    at java.base/java.security.Security.getImpl(Security.java:749)
    at java.base/java.security.KeyStore.getInstance(KeyStore.java:868)
    ... 24 more

This works with Kafka clients, and apparently it also should work with the Confluent Schema Registry according to this PR, so I was hoping that it would also work with Apicurio, but it doesn't look like it.