代码之家  ›  专栏  ›  技术社区  ›  Giorgos Myrianthous

启用SSL加密和身份验证后,Kafka Avro控制台使用者不工作

confluent-schema-registry kafka-consumer-api apache-kafka ssl
  •  0
  • Giorgos Myrianthous  · 技术社区  · 7 年前

    我最近在我的Kafka集群上启用了双向身份验证。

    我正在尝试使用

    kafka-avro-console-consumer --bootstrap-server kafka-host:9092 --topic my_topic --consumer.config client-ssl.properties --from-beginning --property schema.registry.url="https://kafka-host:8081"

    其中的内容 client-ssl.properties 文件是 

    bootstrap.servers=kafka-host:9092
security.protocol=SSL
ssl.truststore.location=/var/private/ssl/clienttruststore.jks
ssl.truststore.password=mypass
ssl.keystore.location=/var/private/ssl/clientkeystore.jks
ssl.keystore.password=mypass
ssl.key.password=mypass

    但会引发以下错误: 

    Processed a total of 1 messages
[2018-12-10 10:22:34,275] ERROR Unknown error when running consumer:  (kafka.tools.ConsoleConsumer$:76)
org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 122
Caused by: javax.net.ssl.SSLHandshakeException: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
        at sun.security.ssl.Alerts.getSSLException(Alerts.java:192)
        at sun.security.ssl.SSLSocketImpl.fatal(SSLSocketImpl.java:1959)
        at sun.security.ssl.Handshaker.fatalSE(Handshaker.java:328)
        at sun.security.ssl.Handshaker.fatalSE(Handshaker.java:322)
        at sun.security.ssl.ClientHandshaker.serverCertificate(ClientHandshaker.java:1614)
        at sun.security.ssl.ClientHandshaker.processMessage(ClientHandshaker.java:216)
        at sun.security.ssl.Handshaker.processLoop(Handshaker.java:1052)
        at sun.security.ssl.Handshaker.process_record(Handshaker.java:987)
        at sun.security.ssl.SSLSocketImpl.readRecord(SSLSocketImpl.java:1072)
        at sun.security.ssl.SSLSocketImpl.performInitialHandshake(SSLSocketImpl.java:1385)
        at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:1413)
        at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:1397)
        at sun.net.www.protocol.https.HttpsClient.afterConnect(HttpsClient.java:559)
        at sun.net.www.protocol.https.AbstractDelegateHttpsURLConnection.connect(AbstractDelegateHttpsURLConnection.java:185)
        at sun.net.www.protocol.http.HttpURLConnection.getInputStream0(HttpURLConnection.java:1564)
        at sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1492)
        at java.net.HttpURLConnection.getResponseCode(HttpURLConnection.java:480)
        at sun.net.www.protocol.https.HttpsURLConnectionImpl.getResponseCode(HttpsURLConnectionImpl.java:347)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:185)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:229)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:409)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:402)
        at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaByIdFromRegistry(CachedSchemaRegistryClient.java:118)
        at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getBySubjectAndId(CachedSchemaRegistryClient.java:191)
        at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getById(CachedSchemaRegistryClient.java:167)
        at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:121)
        at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:93)
        at io.confluent.kafka.formatter.AvroMessageFormatter.writeTo(AvroMessageFormatter.java:160)
        at io.confluent.kafka.formatter.AvroMessageFormatter.writeTo(AvroMessageFormatter.java:152)
        at kafka.tools.ConsoleConsumer$.process(ConsoleConsumer.scala:116)
        at kafka.tools.ConsoleConsumer$.run(ConsoleConsumer.scala:75)
        at kafka.tools.ConsoleConsumer$.main(ConsoleConsumer.scala:53)
        at kafka.tools.ConsoleConsumer.main(ConsoleConsumer.scala)
Caused by: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
        at sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:397)
        at sun.security.validator.PKIXValidator.engineValidate(PKIXValidator.java:302)
        at sun.security.validator.Validator.validate(Validator.java:260)
        at sun.security.ssl.X509TrustManagerImpl.validate(X509TrustManagerImpl.java:324)
        at sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:229)
        at sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:124)
        at sun.security.ssl.ClientHandshaker.serverCertificate(ClientHandshaker.java:1596)
        ... 28 more
Caused by: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
        at sun.security.provider.certpath.SunCertPathBuilder.build(SunCertPathBuilder.java:141)
        at sun.security.provider.certpath.SunCertPathBuilder.engineBuild(SunCertPathBuilder.java:126)
        at java.security.cert.CertPathBuilder.build(CertPathBuilder.java:280)
        at sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:392)
        ... 34 more
[2018-12-10 10:22:34,275] ERROR Unknown error when running consumer:  (kafka.tools.ConsoleConsumer$:76)
org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 122
Caused by: javax.net.ssl.SSLHandshakeException: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
        at sun.security.ssl.Alerts.getSSLException(Alerts.java:192)
        at sun.security.ssl.SSLSocketImpl.fatal(SSLSocketImpl.java:1959)
        at sun.security.ssl.Handshaker.fatalSE(Handshaker.java:328)
        at sun.security.ssl.Handshaker.fatalSE(Handshaker.java:322)
        at sun.security.ssl.ClientHandshaker.serverCertificate(ClientHandshaker.java:1614)
        at sun.security.ssl.ClientHandshaker.processMessage(ClientHandshaker.java:216)
        at sun.security.ssl.Handshaker.processLoop(Handshaker.java:1052)
        at sun.security.ssl.Handshaker.process_record(Handshaker.java:987)
        at sun.security.ssl.SSLSocketImpl.readRecord(SSLSocketImpl.java:1072)
        at sun.security.ssl.SSLSocketImpl.performInitialHandshake(SSLSocketImpl.java:1385)
        at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:1413)
        at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:1397)
        at sun.net.www.protocol.https.HttpsClient.afterConnect(HttpsClient.java:559)
        at sun.net.www.protocol.https.AbstractDelegateHttpsURLConnection.connect(AbstractDelegateHttpsURLConnection.java:185)
        at sun.net.www.protocol.http.HttpURLConnection.getInputStream0(HttpURLConnection.java:1564)
        at sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1492)
        at java.net.HttpURLConnection.getResponseCode(HttpURLConnection.java:480)
        at sun.net.www.protocol.https.HttpsURLConnectionImpl.getResponseCode(HttpsURLConnectionImpl.java:347)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:185)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:229)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:409)
        at io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:402)
        at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaByIdFromRegistry(CachedSchemaRegistryClient.java:118)
        at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getBySubjectAndId(CachedSchemaRegistryClient.java:191)
        at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getById(CachedSchemaRegistryClient.java:167)
        at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:121)
        at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:93)
        at io.confluent.kafka.formatter.AvroMessageFormatter.writeTo(AvroMessageFormatter.java:160)
        at io.confluent.kafka.formatter.AvroMessageFormatter.writeTo(AvroMessageFormatter.java:152)
        at kafka.tools.ConsoleConsumer$.process(ConsoleConsumer.scala:116)
        at kafka.tools.ConsoleConsumer$.run(ConsoleConsumer.scala:75)
        at kafka.tools.ConsoleConsumer$.main(ConsoleConsumer.scala:53)
        at kafka.tools.ConsoleConsumer.main(ConsoleConsumer.scala)
Caused by: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
        at sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:397)
        at sun.security.validator.PKIXValidator.engineValidate(PKIXValidator.java:302)
        at sun.security.validator.Validator.validate(Validator.java:260)
        at sun.security.ssl.X509TrustManagerImpl.validate(X509TrustManagerImpl.java:324)
        at sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:229)
        at sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:124)
        at sun.security.ssl.ClientHandshaker.serverCertificate(ClientHandshaker.java:1596)
        ... 28 more
Caused by: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
        at sun.security.provider.certpath.SunCertPathBuilder.build(SunCertPathBuilder.java:141)
        at sun.security.provider.certpath.SunCertPathBuilder.engineBuild(SunCertPathBuilder.java:126)
        at java.security.cert.CertPathBuilder.build(CertPathBuilder.java:280)
        at sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:392)
        ... 34 more

    我应该如何解决这个问题?

    在启用模式注册表的双向认证之后,我确定了与Java卡夫卡用户相同的问题,并在运行消费者时包括以下参数: .jar 文件: 

    -Djavax.net.ssl.trustStore=/var/private/ssl/clienttruststore.jks 
-Djavax.net.ssl.trustStorePassword=mypass
-Djavax.net.ssl.keyStore=/var/private/ssl/clientkeystore.jks 
-Djavax.net.ssl.keyStorePassword=mypass
    1 回复  |  直到 7 年前
        1
  •  0
  •   Giorgos Myrianthous    7 年前

    为了解决这个问题,我必须导出以下环境变量: 

    export SCHEMA_REGISTRY_OPTS="-Djavax.net.ssl.keyStore=/var/private/ssl/clientkeystore.jks -Djavax.net.ssl.trustStore=/var/private/ssl/clienttruststore.jks -Djavax.net.ssl.keyStorePassword=mypass-Djavax.net.ssl.trustStorePassword=mypass"
export KAFKA_OPTS="-Djavax.net.ssl.keyStore=/var/private/ssl/clientkeystore.jks -Djavax.net.ssl.trustStore=/var/private/ssl/clienttruststore.jks -Djavax.net.ssl.keyStorePassword=mypass-Djavax.net.ssl.trustStorePassword=mypass"

    对于更持久的解决方案,您可以在 ~/.bashrc 文件或 /etc/environment (对于后者,不包括 export ,只是 KAFKA_OPTS=.. 和 SCHEMA_REGISTRY_OPTS=.. )

    推荐文章
    关于   移动版
    代码之家 - 一站式码农服务社区
    沪ICP备11025650号