spring-boot - Spring Cloud Kafka Stream 无法创建生产者配置错误

标签 spring-boot gradle kotlin spring-cloud-stream spring-kafka

我有两个带有 Kafka-stream 依赖项的 Spring Boot 项目，它们在 gradle 中具有完全相同的依赖项和完全相同的配置，但其中一个项目在启动时记录错误如下

11:35:37.974 [restartedMain] INFO  o.a.k.c.admin.AdminClientConfig - AdminClientConfig values: 
    bootstrap.servers = [192.169.0.109:6667]
    client.id = client
    connections.max.idle.ms = 300000
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    receive.buffer.bytes = 65536
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 120000
    retries = 5
    retry.backoff.ms = 100
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    send.buffer.bytes = 131072
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = null
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLS
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS

11:35:38.017 [restartedMain] INFO  o.a.kafka.common.utils.AppInfoParser - Kafka version : 1.0.0
11:35:38.017 [restartedMain] INFO  o.a.kafka.common.utils.AppInfoParser - Kafka commitId : aaa7af6d4a11b29d
11:35:38.136 [restartedMain] INFO  o.s.c.s.b.k.p.KafkaTopicProvisioner - Using kafka topic for outbound: createschedule
11:36:08.147 [restartedMain] ERROR o.s.c.stream.binding.BindingService - Failed to create producer binding; retrying in 30 seconds
org.springframework.cloud.stream.provisioning.ProvisioningException: provisioning exception; nested exception is java.util.concurrent.TimeoutException
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopic(KafkaTopicProvisioner.java:243)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.provisionProducerDestination(KafkaTopicProvisioner.java:126)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.provisionProducerDestination(KafkaTopicProvisioner.java:71)
    at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindProducer(AbstractMessageChannelBinder.java:140)
    at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindProducer(AbstractMessageChannelBinder.java:77)
    at org.springframework.cloud.stream.binder.AbstractBinder.bindProducer(AbstractBinder.java:136)
    at org.springframework.cloud.stream.binding.BindingService.doBindProducer(BindingService.java:244)
    at org.springframework.cloud.stream.binding.BindingService.bindProducer(BindingService.java:221)
    at org.springframework.cloud.stream.binding.BindableProxyFactory.bindOutputs(BindableProxyFactory.java:252)
    at org.springframework.cloud.stream.binding.OutputBindingLifecycle.doStartWithBindable(OutputBindingLifecycle.java:46)
    at java.util.Iterator.forEachRemaining(Iterator.java:116)
    at java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1801)
    at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:580)
    at org.springframework.cloud.stream.binding.AbstractBindingLifecycle.start(AbstractBindingLifecycle.java:47)
    at org.springframework.cloud.stream.binding.OutputBindingLifecycle.start(OutputBindingLifecycle.java:29)
    at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:181)
    at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:52)
    at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:356)
    at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:157)
    at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:121)
    at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:884)
    at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.finishRefresh(ServletWebServerApplicationContext.java:161)
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:552)
    at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:140)
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:752)
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:388)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:327)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1246)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1234)
    at com.und.ClientPanelApplicationKt.main(ClientPanelApplication.kt:21)
    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 org.springframework.boot.devtools.restart.RestartLauncher.run(RestartLauncher.java:49)
Caused by: java.util.concurrent.TimeoutException: null
    at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:108)
    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:225)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopicAndPartitions(KafkaTopicProvisioner.java:271)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopicIfNecessary(KafkaTopicProvisioner.java:251)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopic(KafkaTopicProvisioner.java:236)
    ... 34 common frames omitted

我正在使用 kotlin 2.30 和 Spring Release 2.0.0.0 我的构建 gradle 的相关部分如下

ext {
    kotlinVersion = '1.2.30'
    springBootVersion = '2.0.0.RELEASE'
    springCloudVersion = 'Finchley.M8'
}

dependencies{
  compile('org.springframework.cloud:spring-cloud-starter-stream-kafka')
}

我已经定义了一个输出 channel 如下 

interface EventStream {

    @Output("createschedule")
    fun outputEvent(): MessageChannel

}

下面是我的代码的相关部分，我在其中定义了发送者和监听器

@Service
class TestService {



    @Autowired
    private lateinit var eventStream: EventStream

    fun test() {
        processSchedule("Hello")
    }






    @SendTo("createschedule")
    fun processSchedule(campaign: String): String {
        return campaign
    }

    @StreamListener("createschedule")
    fun listenSchedule(campaign: String) {
        println(campaign)
        //return campaign
    }

}

以下是我的 application.yaml 的相关部分 

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: ${KAFKA_IP}
          defaultBrokerPort: ${KAFKA_PORT}
          zkNodes: ${ZK_IP}
          defaultZkPort: ${ZK_PORT}
        bindings:
          createschedule:
            group: createscheduleGroup
            destination: createschedule
            consumer:
              group: createscheduleGroup



  kafka:
    admin:
      properties:
        #security.protocol: SSL
        client.id: client

正如我已经说过的，它在启动时会抛出错误，在启动后它会在日志中不断抛出错误

11:54:38.231 [pool-3-thread-1] INFO  o.s.c.s.b.k.p.KafkaTopicProvisioner - Using kafka topic for outbound: createschedule
11:54:59.070 [kafka-admin-client-thread | client-panel] WARN  o.apache.kafka.clients.NetworkClient - [AdminClient clientId=client-panel] Connection to node -1 could not be established. Broker may not be available.
11:55:08.234 [pool-3-thread-1] ERROR o.s.c.stream.binding.BindingService - Failed to create producer binding; retrying in 30 seconds
org.springframework.cloud.stream.provisioning.ProvisioningException: provisioning exception; nested exception is java.util.concurrent.TimeoutException
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopic(KafkaTopicProvisioner.java:243)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.provisionProducerDestination(KafkaTopicProvisioner.java:126)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.provisionProducerDestination(KafkaTopicProvisioner.java:71)
    at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindProducer(AbstractMessageChannelBinder.java:140)
    at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindProducer(AbstractMessageChannelBinder.java:77)
    at org.springframework.cloud.stream.binder.AbstractBinder.bindProducer(AbstractBinder.java:136)
    at org.springframework.cloud.stream.binding.BindingService.lambda$rescheduleProducerBinding$2(BindingService.java:262)
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
    at java.util.concurrent.Executors$RunnableAdapter.call$$$capture(Executors.java:511)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java)
    at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)
    at java.util.concurrent.FutureTask.run(FutureTask.java)
    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:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.util.concurrent.TimeoutException: null
    at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:108)
    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:225)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopicAndPartitions(KafkaTopicProvisioner.java:271)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopicIfNecessary(KafkaTopicProvisioner.java:251)
    at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopic(KafkaTopicProvisioner.java:236)
    ... 16 common frames omitted

最佳答案

这是卡夫卡的问题。 如果您使用的是 docker cointainer，则停止 kafka，删除 kafka 并启动 kafka cointainer。

关于spring-boot - Spring Cloud Kafka Stream 无法创建生产者配置错误，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/49421979/

上一篇：kotlin - 如何将 Kotlin 代码引用从函数更改为 intellij 插件中的字段？

下一篇：types - 如何将数据类型映射到该数据类型的集合

相关文章：

java - 如何将多列与其他列一起以 JSON 形式返回

spring-boot - 从 SpringBoot 2.6.6 升级到 2.6.7 更改了验证行为(javax.validation.ConstraintDeclarationException : HV000151 )

spring-boot - 如何使用证书将azure Ad与java Web应用程序连接？

java - @Async 不适用于返回类型为 void 的方法

java - 需要访问源的 Gradle 自定义插件

java - 从单个源文件夹 Gradle 多个 jar

java - 将我的Kotlin Maven发布代码转换为Java Gradle插件

generics - Kotlin 注释作为通用函数参数

kotlin - 如果条件在kotlin中的数据类构造函数中

java - 如何在没有双重请求的情况下使用 Google Places API 获取特定类型的 Lat Longs 列表？

©2024 IT工具网  联系我们