0

我的应用程序使用来自一台机器上运行的 Kafka 服务器的消息，然后将它们转发到另一个在其他实例上运行的远程 Kafka。在我将应用程序部署到 Cloud Foundry 并向第一台 Kafka 服务器发送消息后，应用程序按预期工作。消息被消费并转发到远程 Kafka。

但是，在那之后，我在 Cloud Foundry 中得到了以下异常的无限循环（在我的本地机器上也以较慢的速度）：

堆栈跟踪：

Fri Jun 03 2016 12:20:34 GMT-0600 (Mountain Daylight Time) [App/0] OUT 2016-06-03 18:20:34.900 WARN 29 --- [ad | producer-1] o.apache.kafka.common.network.Selector : Error in I/O with localhost/127.0.0.1
Fri Jun 03 2016 12:20:34 GMT-0600 (Mountain Daylight Time) [App/0] OUT at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) ~[na:1.8.0_65-]
Fri Jun 03 2016 12:20:34 GMT-0600 (Mountain Daylight Time) [App/0] OUT java.net.ConnectException: Connection refused
Fri Jun 03 2016 12:20:34 GMT-0600 (Mountain Daylight Time) [App/0] OUT at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717) ~[na:1.8.0_65-]
Fri Jun 03 2016 12:20:34 GMT-0600 (Mountain Daylight Time) [App/0] OUT at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:122) [kafka-clients-0.8.2.2.jar!/:na]
Fri Jun 03 2016 12:20:34 GMT-0600 (Mountain Daylight Time) [App/0] OUT at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:191) [kafka-clients-0.8.2.2.jar!/:na]
Fri Jun 03 2016 12:20:34 GMT-0600 (Mountain Daylight Time) [App/0] OUT at org.apache.kafka.common.network.Selector.poll(Selector.java:238) ~[kafka-clients-0.8.2.2.jar!/:na]
Fri Jun 03 2016 12:20:34 GMT-0600 (Mountain Daylight Time) [App/0] OUT at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:192) [kafka-clients-0.8.2.2.jar!/:na]
Fri Jun 03 2016 12:20:34 GMT-0600 (Mountain Daylight Time) [App/0] OUT at java.lang.Thread.run(Thread.java:745) [na:1.8.0_65-]

我的应用程序 yaml 文件是这样的

应用 YML：

spring:
  cloud:
    stream:
      bindings:
        activationMsgQueue:
          binder: kafka1
          destination: test
          contentType: application/json
          consumer:
            resetOffsets: true
            startOffset: latest
        input:
          binder: kafka2
          content-type: application/x-java-object;type=com.comcast.activation.message.vo.ActivationDataInfo
          destination: test
          group: prac  
      binders:
        kafka1:
          type: kafka
          environment:
            spring:
              kafka:
                host: caapmsg-as-a1p.sys.comcast.net
        kafka2:
          type: kafka
          environment:
            spring:
              kafka:
                host: caapmsg-as-a3p.sys.comcast.net
      default-binder: kafka2                    
      kafka:
        binder:
          zk-nodes: caapmsg-as-a1p.sys.comcast.net, caapmsg-as-a3p.sys.comcast.net

我观察到，如果我包含下面的配置，错误就会消失，但现在我有一个无限循环的消息被消耗和发送。

片段： 

kafka:
        binder:
           brokers: caapmsg-as-a1p.sys.comcast.net, caapmsg-as-a3p.sys.comcast.net
          zk-nodes: caapmsg-as-a1p.sys.comcast.net, caapmsg-as-a3p.sys.comcast.net

我需要做什么来停止这个无限循环？

嗨 Marius， 感谢您回复 SOS 电话。我对上述问题进行了改进。流程现在从 a1p(topic:test) 消费，如果消息有效，则转发到 a3p(topic:test)，否则将错误消息发送到 a1p(topic:errorMsgQueue)。我有以下应用程序。.yml 文件

弹簧：云：流：绑定：errorMsgQueue：binder：kafka1 目的地：errorMsgQueue contentType：application/json 输入：binder：kafka2 内容类型：application/x-java-object;type=com.comcast.activation.message.vo。 ActivationDataInfo 目的地：测试组：prac
activationMsgQueue: binder: kafka3 目的地: 测试 contentType: application/json binders: kafka1: type: kafka environment: spring: cloud: stream: kafka: binder: brokers: caapmsg-as-a1p.sys.comcast.net zk-nodes: caapmsg -as-a1p.sys.comcast.net kafka2: 类型: kafka 环境: spring: cloud: stream: kafka: binder: brokers: caapmsg-as-a3p.sys.comcast.net zk-nodes: caapmsg-as-a3p。 sys.comcast.net kafka3: 类型: kafka 环境: spring: cloud:流：kafka：binder：代理：caapmsg-as-a1p.sys.comcast.net zk-nodes：caapmsg-as-a1p.sys.comcast.net 默认绑定器：kafka2

我仍然得到一个无限循环。我究竟做错了什么？

4

1 回答 1

0

spring.kafka.host不是 Spring Cloud Stream 的有效配置选项。http://docs.spring.io/spring-cloud-stream/docs/1.0.0.RELEASE/reference/htmlsingle/index.html#_kafka_binder_properties是活页夹支持的唯一属性。

此外，您的应用程序似乎正在混合两个集群的配置。（我假设它们是单独的集群？）

它应该是这样的：

spring: cloud: stream: bindings: activationMsgQueue: binder: kafka1 destination: test contentType: application/json consumer: resetOffsets: true startOffset: latest input: binder: kafka2 content-type: application/x-java-object;type=com.comcast.activation.message.vo.ActivationDataInfo destination: test group: prac
binders: kafka1: type: kafka environment: spring: cloud: stream: kafka: binder: brokers: caapmsg-as-a1p.sys.comcast.net zk-nodes: caapmsg-as-a1p.sys.comcast.net kafka2: type: kafka environment: spring: cloud: stream: kafka: binder: brokers: caapmsg-as-a3p.sys.comcast.net zk-nodes: caapmsg-as-a3p.sys.comcast.net default-binder: kafka2

有关详细信息，请参阅此示例https://github.com/spring-cloud/spring-cloud-stream-samples/blob/master/multibinder-differentsystems/src/main/resources/application.yml。

我怀疑无限循环是由向同一主题发送和接收消息引起的。

于 2016-06-07T02:58:42.417 回答