问题：
Spring cloud Stream：具有单独jaas配置的Kafka生产者和消费者的多个绑定器不能一起工作
宣瀚
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
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
共有1个答案
许展鹏
您的jaas配置不正确；应该是
jaas:
config: com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="\src\main\keytab\XXXXXCON.keytab" principal="XXXXXCON@tt.com" doNotPrompt=true refreshKrb5Config=true;
请参阅文档。
我无法解释为什么它在你只有一个活页夹的情况下完全起作用。
...
input:
consumer:
useNativeDecoding: true
...
output:
producer:
useNativeEncoding: true
类似资料：
-
我有一个使用ActiveMQ的消息队列。web请求用persistency=true将消息放入队列。现在，我有两个消费者，它们都作为单独的会话连接到这个队列。使用者1总是确认消息，但使用者2从不这样做。 JMS队列实现负载平衡器语义。一条消息将被一个使用者接收。如果在发送消息时没有可用的使用者，它将被保留，直到有可以处理消息的使用者可用为止。如果使用者接收到一条消息，但在关闭之前没有确认它，那么该
-
在这种情况下，我是否需要求助于Kafka事务API来在消费者轮询循环中创建事务生产者，在该循环中，我在事务中执行：（1）处理消耗的记录和（2）在关闭事务之前提交它们的偏移量。在这种情况下，普通的commitsync/commitasync是否有效？
-
我正在创建一个系统，其中前端服务将消息推送到Kafka请求主题，并为一些下游后端消费者（实际上是一个最终推送回Kafka的复杂系统）监听另一个响应主题，以处理请求消息并最终推进到“回应”话题。 我试图找出最优雅的方法来确保消费者监听适当的分区并收到响应，并且后端推送到前端消费者正在监听的分区。我们总是需要确保响应到达产生初始消息的同一个消费者。 到目前为止，我有两种解决方案，但都不是特别令人满意的
-
我有三根线。线程1(T1)是生成器，它生成数据。线程2和线程3(T2和T3)分别等待T1的数据在单独的循环中处理。我正在考虑在线程之间共享BlockingQueue,并通过调用“Take”让T2和T3等待。
-
我创建了两个apache camel（blueprint XML）kafka项目，一个是kafka-producer（接受请求并将其存储在kafka服务器中），另一个是kafka-consumer（从kafka服务器获取ups消息并处理它们）。 这个设置对单个主题和单个消费者都很有效。然而，我如何在同一个Kafka主题中创建单独的消费者组？如何在不同的消费者群体中路由同一主题中的多个消费者特定消息
-
我正在与Java一起研究生产者-消费者问题的多生产者和消费者用例。代码在GitHub上。同样的实现适用于单个生产者和消费者用例，但对于多生产者和消费者用例却表现得很奇怪。 我有一些关于输出的问题： 一开始,所有生产者和一个消费者都有锁： 我想所有的线程都应该竞争锁，并且最多应该有一个线程拥有所有时间的锁？是不是所有的制作人都共用这个锁？当生产者线程t1持有锁时，使用者线程t5是如何获得锁的？ 它运