无法使用ssl配置创建多个kafka绑定器

rkttyhzu  于 2021-06-04  发布在  Kafka
关注(0)|答案(0)|浏览(372)

我正在尝试通过sasl\u ssl协议和jaas config连接到kafka群集，如下所示：

spring:
  cloud:
    stream:
      bindings:
        binding-1:
          binder: kafka-1-with-ssl
          destination: <destination-1>
          content-type: text/plain
          group: <group-id-1>
          consumer:
            header-mode: headers
        binding-2:
          binder: kafka-2-with-ssl
          destination: <destination-2>
          content-type: text/plain
          group: <group-id-2>
          consumer:
            header-mode: headers

      binders:
        kafka-1-with-ssl:
          type: kafka
          defaultCandidate: false
          environment:
            spring:
              cloud:
                stream:
                  kafka:
                    binder: 
                      brokers: <broker-hostnames-1>
                      configuration:
                        ssl:
                          truststore:
                            location: <location-1>
                            password: <ts-password-1>
                            type: JKS
                      jaas:
                        loginModule: org.apache.kafka.common.security.scram.ScramLoginModule
                        options:
                          username: <username-1>
                          password: <password-1>

        kafka-2-with-ssl:
          type: kafka
          defaultCandidate: false
          environment:
            spring:
              cloud:
                stream:
                  kafka:
                    binder: 
                      brokers: <broker-hostnames-2>
                      configuration:
                        ssl:
                          truststore:
                            location: <location-2>
                            password: <ts-password-2>
                            type: JKS
                      jaas:
                        loginModule: org.apache.kafka.common.security.scram.ScramLoginModule
                        options:
                          username: <username-2>
                          password: <password-2>
      kafka:
        binder:
          configuration:
            security:
              protocol: SASL_SSL
            sasl:
              mechanism: SCRAM-SHA-256

上面的配置与springcloudstream的官方gitrepo上提供的示例配置是内联的。
在图书馆的gitrepo上提出的类似问题表示，它在最新版本中得到了修复，但看起来并非如此。获取以下错误：
springbootversion:2.2.8和SpringCloudStreamDependencies版本-horsham.sr6。

Failed to create consumer binding; retrying in 30 seconds | org.springframework.cloud.stream.binder.BinderException: Exception thrown while starting consumer: 
    at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:461)
    at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:90)
    at org.springframework.cloud.stream.binder.AbstractBinder.bindConsumer(AbstractBinder.java:143)
    at org.springframework.cloud.stream.binding.BindingService.lambda$rescheduleConsumerBinding$1(BindingService.java:201)
    at org.springframework.cloud.sleuth.instrument.async.TraceRunnable.run(TraceRunnable.java:68)
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java)
    at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.common.KafkaException: Failed to create new KafkaAdminClient
    at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:407)
    at org.apache.kafka.clients.admin.AdminClient.create(AdminClient.java:65)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createAdminClient(KafkaTopicProvisioner.java:246)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.doProvisionConsumerDestination(KafkaTopicProvisioner.java:216)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.provisionConsumerDestination(KafkaTopicProvisioner.java:183)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.provisionConsumerDestination(KafkaTopicProvisioner.java:79)
    at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:402)
    ... 12 common frames omitted
Caused by: org.apache.kafka.common.KafkaException: javax.security.auth.login.LoginException: KrbException: Cannot locate default realm
    at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:160)
    at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:146)
    at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:67)
    at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:99)
    at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:382)
    ... 18 common frames omitted
Caused by: javax.security.auth.login.LoginException: KrbException: Cannot locate default realm
    at com.sun.security.auth.module.Krb5LoginModule.attemptAuthentication(Krb5LoginModule.java:804)
    at com.sun.security.auth.module.Krb5LoginModule.login(Krb5LoginModule.java:617)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at javax.security.auth.login.LoginContext.invoke(LoginContext.java:755)
    at javax.security.auth.login.LoginContext.access$000(LoginContext.java:195)
    at javax.security.auth.login.LoginContext$4.run(LoginContext.java:682)
    at javax.security.auth.login.LoginContext$4.run(LoginContext.java:680)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.login.LoginContext.invokePriv(LoginContext.java:680)
    at javax.security.auth.login.LoginContext.login(LoginContext.java:587)
    at org.apache.kafka.common.security.authenticator.AbstractLogin.login(AbstractLogin.java:60)
    at org.apache.kafka.common.security.authenticator.LoginManager.<init>(LoginManager.java:61)
    at org.apache.kafka.common.security.authenticator.LoginManager.acquireLoginManager(LoginManager.java:111)
    at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:149)
    ... 22 common frames omitted
Caused by: sun.security.krb5.RealmException: KrbException: Cannot locate default realm
    at sun.security.krb5.Realm.getDefault(Realm.java:68)
    at sun.security.krb5.PrincipalName.<init>(PrincipalName.java:462)
    at sun.security.krb5.PrincipalName.<init>(PrincipalName.java:471)
    at com.sun.security.auth.module.Krb5LoginModule.attemptAuthentication(Krb5LoginModule.java:706)
    ... 38 common frames omitted
Caused by: sun.security.krb5.KrbException: Cannot locate default realm
    at sun.security.krb5.Config.getDefaultRealm(Config.java:1029)
    at sun.security.krb5.Realm.getDefault(Realm.java:64)
    ... 41 common frames omitted

这让我觉得库没有正确地获取配置道具，因为 jaas.loginModule 指定为 ScramLoginModule 但它在使用 Krb5LoginModule 进行身份验证。
但是，令人吃惊的是，当按照以下方式进行配置时（区别在于最后一部分在binder的环境之外使用ssl凭据），它连接到全局ssl props（在binder的env之外）中指定的binder，并且默默地忽略另一个binder，而不显示任何错误日志。
说明绑定器的密码凭据 kafka-2-with-ssl 如果在全局ssl props中指定了，则将创建该绑定器，并且订阅该绑定器的绑定将开始使用事件。但这仅在需要创建单个绑定器时才有用。

spring:
  cloud:
    stream:
      bindings:
        binding-1:
          binder: kafka-1-with-ssl
          destination: <destination-1>
          content-type: text/plain
          group: <group-id-1>
          consumer:
            header-mode: headers
        binding-2:
          binder: kafka-2-with-ssl
          destination: <destination-2>
          content-type: text/plain
          group: <group-id-2>
          consumer:
            header-mode: headers

      binders:
        kafka-1-with-ssl:
          type: kafka
          defaultCandidate: false
          environment:
            spring:
              cloud:
                stream:
                  kafka:
                    binder: 
                      brokers: <broker-hostnames-1>
                      configuration:
                        ssl:
                          truststore:
                            location: <location-1>
                            password: <ts-password-1>
                            type: JKS
                      jaas:
                        loginModule: org.apache.kafka.common.security.scram.ScramLoginModule
                        options:
                          username: <username-1>
                          password: <password-1>

        kafka-2-with-ssl:
          type: kafka
          defaultCandidate: false
          environment:
            spring:
              cloud:
                stream:
                  kafka:
                    binder: 
                      brokers: <broker-hostnames-2>
                      configuration:
                        ssl:
                          truststore:
                            location: <location-2>
                            password: <ts-password-2>
                            type: JKS
                      jaas:
                        loginModule: org.apache.kafka.common.security.scram.ScramLoginModule
                        options:
                          username: <username-2>
                          password: <password-2>
      kafka:
        binder:
          configuration:
            security:
              protocol: SASL_SSL
            sasl:
              mechanism: SCRAM-SHA-256 
            ssl:
              truststore:
                location: <location-2>
                password: <ts-password-2> 
                type: JKS
          jaas:
            loginModule: org.apache.kafka.common.security.scram.ScramLoginModule
             options:
               username: <username-2>
               password: <password-2>

向您保证ssl凭据没有问题。用ssl-kafka绑定器中的任何一个进行了认真的测试，成功地单独创建了。其目的是通过sasl\u ssl协议连接到多个kafka绑定器。提前谢谢。

springapache-kafkaspring-bootspring-cloud-streamspring-cloud-stream-binder-kafka

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com