我正在尝试使用 spring cloud 和 binder 在同一个 Spring boot 应用程序中实现 Kafka 消费者和 Kafka 生产者。如果单独执行，两者都可以成功运行，但如果一起执行，则只有 Kafka Producer 能够成功连接到 kafka 集群，但 Kafka Consumer 无法登录到 Kafka 集群。我认为问题在于 kafka 生产者和消费者的多个/不同的 jaas 配置。请在我的 application.yml 文件下面找到
spring:
cloud:
stream:
bindings:
input:
group: consumer-tt1
useNativeEncoding: true
destination: consumer-topic
content-type: application/json
binder: consumer
output:
destination: producer-topic
useNativeEncoding: true
content-type: application/*+avro
binder: producer
binders:
consumer:
type: kafka
environment:
spring:
cloud:
stream:
kafka:
binder:
autoCreateTopics: false
autoAddPartitions: false
consumer-properties:
key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
value.deserializer: org.apache.kafka.common.serialization.StringDeserializer
brokers: xxxxx.tt.com:9092
jaas:
loginModule: com.sun.security.auth.module.Krb5LoginModule
controlFlag: required
options:
useKeyTab: true
storeKey: true
keyTab: \src\main\keytab\XXXXXCON.keytab
principal: XXXXXCON@tt.com
doNotPrompt: true
refreshKrb5Config: true
configuration:
application: XXXXXCON
sasl:
kerberos:
realm: tt.com
kdc: tt.com
service:
name: kafka
jaas:
loginModule: com.sun.security.auth.module.Krb5LoginModule
controlFlag: required
config:
useKeyTab: true
storeKey: true
keyTab: \src\main\keytab\XXXXXCON.keytab
principal: XXXXXCON@tt.com
doNotPrompt: true
refreshKrb5Config: true
security:
protocol: SASL_SSL
ssl:
truststore:
location: \src\main\keytab\truststore.jks
password: 123456789
type: JKS
producer:
type: kafka
environment:
spring:
cloud:
stream:
kafka:
binder:
autoCreateTopics: false
autoAddPartitions: false
producer-properties:
key.serializer: org.apache.kafka.common.serialization.StringSerializer
value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
schema.registry.url: http:/kafka-schema:8484
brokers: xxxxx.tt.com:9092
jaas:
loginModule: com.sun.security.auth.module.Krb5LoginModule
controlFlag: required
options:
useKeyTab: true
storeKey: true
keyTab: \src\main\keytab\XXXXXPRO.keytab
principal: XXXXXPRO@tt.com
doNotPrompt: true
refreshKrb5Config: true
configuration:
application:
id: XXXXXPRO
sasl:
kerberos:
realm: tt.com
kdc: tt.com
service:
name: kafka
jaas:
loginModule: com.sun.security.auth.module.Krb5LoginModule
controlFlag: required
config:
useKeyTab: true
storeKey: true
keyTab: \src\main\keytab\XXXXXPRO.keytab
principal: XXXXXPRO@tt.com
doNotPrompt: true
refreshKrb5Config: true
security:
protocol: SASL_SSL
ssl:
truststore:
location: \src\main\keytab\truststore.jks
password: 123456789
type: JKS
schema-registry-client:
endpoint: http:/kafka-schema:8484
如果我在主类中使用@EnableBinding(Source.class) 或@EnableBinding(Sink.class) 运行相同的application.yml，它会以Kafka Producer 或Kafka Consumer 的身份成功连接kafka 集群。但是当我使用@EnableBinding（Processor.class）运行相同的application.yml时，我遇到了kafka消费者的错误（Kafka Producer工作正常并连接到Kafka集群）。问题仅与 Kafka 消费者有关：无权访问该主题。
org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [consumer-topic]
2020-03-24 19:45:07.794 WARN 19000 --- [ main] o.s.c.s.b.k.p.KafkaTopicProvisioner : No partitions have been retrieved for the topic (consumer-topic). This will affect the health check.
2020-03-24 19:45:07.794 WARN 19000 --- [ main] o.s.c.s.b.k.p.KafkaTopicProvisioner : The number of expected partitions was: 1, but 0 has been found instead.There will be 1 idle consumers
2020-03-24 19:45:07.796 ERROR 19000 --- [ main] o.s.cloud.stream.binding.BindingService : 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:435) ~[spring-cloud-stream-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:97) ~[spring-cloud-stream-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.cloud.stream.binder.AbstractBinder.bindConsumer(AbstractBinder.java:142) ~[spring-cloud-stream-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.cloud.stream.binding.BindingService.doBindConsumer(BindingService.java:144) [spring-cloud-stream-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.cloud.stream.binding.BindingService.bindConsumer(BindingService.java:122) [spring-cloud-stream-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.cloud.stream.binding.BindableProxyFactory.createAndBindInputs(BindableProxyFactory.java:254) [spring-cloud-stream-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.cloud.stream.binding.InputBindingLifecycle.doStartWithBindable(InputBindingLifecycle.java:58) [spring-cloud-stream-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at java.util.LinkedHashMap$LinkedValues.forEach(LinkedHashMap.java:608) ~[na:1.8.0_162]
at org.springframework.cloud.stream.binding.AbstractBindingLifecycle.start(AbstractBindingLifecycle.java:48) ~[spring-cloud-stream-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.cloud.stream.binding.InputBindingLifecycle.start(InputBindingLifecycle.java:34) [spring-cloud-stream-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:182) ~[spring-context-5.1.9.RELEASE.jar:5.1.9.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:53) ~[spring-context-5.1.9.RELEASE.jar:5.1.9.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:360) ~[spring-context-5.1.9.RELEASE.jar:5.1.9.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:158) ~[spring-context-5.1.9.RELEASE.jar:5.1.9.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:122) ~[spring-context-5.1.9.RELEASE.jar:5.1.9.RELEASE]
at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:893) ~[spring-context-5.1.9.RELEASE.jar:5.1.9.RELEASE]
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.finishRefresh(ServletWebServerApplicationContext.java:162) ~[spring-boot-2.1.8.RELEASE.jar:2.1.8.RELEASE]
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:552) ~[spring-context-5.1.9.RELEASE.jar:5.1.9.RELEASE]
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:141) ~[spring-boot-2.1.8.RELEASE.jar:2.1.8.RELEASE]
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:744) ~[spring-boot-2.1.8.RELEASE.jar:2.1.8.RELEASE]
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:391) ~[spring-boot-2.1.8.RELEASE.jar:2.1.8.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:312) ~[spring-boot-2.1.8.RELEASE.jar:2.1.8.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1215) ~[spring-boot-2.1.8.RELEASE.jar:2.1.8.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1204) ~[spring-boot-2.1.8.RELEASE.jar:2.1.8.RELEASE]
at com.rbc.ess.ESSEventTransform.EssEventTransformApplication.main(EssEventTransformApplication.java:21) ~[classes/:na]
Caused by: java.lang.IllegalArgumentException: A list of partitions must be provided
at org.springframework.util.Assert.isTrue(Assert.java:118) ~[spring-core-5.1.9.RELEASE.jar:5.1.9.RELEASE]
at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:446) ~[spring-cloud-stream-binder-kafka-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:133) ~[spring-cloud-stream-binder-kafka-2.1.4.RELEASE.jar:2.1.4.RELEASE]
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:382) ~[spring-cloud-stream-2.1.4.RELEASE.jar:2.1.4.RELEASE]
... 24 common frames omitted
请检查并阐明如何在 Spring Cloud Stream binder application 中传递多个 jaas 配置