o、 阿帕奇。Kafka。常见的网络选择器：localhost/127.0.0.1的I/O错误

Question

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

然而，在这之后，我在Cloud Foundry（以及在我的本地机器上以较慢的速度）中得到了下面的无限循环异常：

StackTrace：

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

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

SNIPPET：

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

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

你好，谢谢你回复求救电话。我对上述问题有一个改进。现在，流将从a1p（主题：test）消费，如果消息有效，则转发到a3p（主题：test），否则将错误消息发送到a1p（主题：errorMsgQueue）。我有下面的申请表。yml文件

spring:cloud:stream:bindings:errorMsgQueue:binder:kafka1目的地：errorMsgQueue contentType:application/json输入：binder:kafka2内容类型：application/x-java-object；type=com。康卡斯特。激活。消息哇。ActivationDataInfo目的地：测试组：prac
activationMsgQueue:binder:kafka3目的地：测试内容类型：应用程序/json绑定器：kafka1:type:kafka环境：spring:cloud:stream:kafka:binder:brokers:caapmsg-as-a1p。系统。康卡斯特。net zk节点：caapmsg-as-a1p。系统。康卡斯特。net kafka2:type:kafka环境：spring:cloud:stream:kafka:binder:brokers:caapmsg-as-a3p。系统。康卡斯特。net zk节点：caapmsg-as-a3p。系统。康卡斯特。net kafka3:type:kafka环境：spring:cloud:stream:kafka:binder:brokers:caapmsg-as-a1p。系统。康卡斯特。net zk节点：caapmsg-as-a1p。系统。康卡斯特。净默认活页夹：kafka2

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

Answer 1

spring。Kafka。主机不是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:resetoffset:true startOffset:latest input:binder:kafka2 content type:application/x-java-object；type=com。康卡斯特。激活。消息哇。ActivationDataInfo目的地：测试组：prac
绑定器：Kafka1：类型：Kafka环境：Spring：云：流：Kafka：绑定器：代理：caapmsg-as-a1p。系统。康卡斯特。net zk节点：caapmsg-as-a1p。系统。康卡斯特。net kafka2:type:kafka环境：spring:cloud:stream:kafka:binder:brokers:caapmsg-as-a3p。系统。康卡斯特。net zk节点：caapmsg-as-a3p。系统。康卡斯特。净默认活页夹：kafka2

看到这个例子了吗https://github.com/spring-cloud/spring-cloud-stream-samples/blob/master/multibinder-differentsystems/src/main/resources/application.yml详细信息。

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