title: o.apache.kafka.common.network.Selector : Error in I/O with localhost/127.0.0.1
body: <p>My application consumes messages from a  Kafka server running  on one  machine and then forwards them to another remote Kafka running on some other instance. After I deploy my app to Cloud Foundry and send a message to the first Kafka server, the application works as expected. The message gets consumed and forwards it to the remote Kafka. </p>
  
  <p>However after that I get a infinite loop of the exception below in Cloud Foundry(and on my local machine as well at a slower pace): </p>
  
  <p>StackTrace: </p>
  
  <pre><code>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-]
  </code></pre>
  
  <p>My application yaml file is like this</p>
  
  <p>Application YML : </p>
  
  <pre><code>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
  </code></pre>
  
  <p>I have observed that if I include the config below, the error disappears, but now I have a infinite loop of messages getting consumed and sent.</p>
  
  <p>SNIPPET:    </p>
  
  <pre><code>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
  </code></pre>
  
  <p>What do I need to do to stop this infinite loop?</p>
  
  <p>Hi Marius, Thank you for responding to the SOS call. I have an enhancement to the above problem. The flow is now to consume from a1p(topic:test), and forward to a3p(topic : test) if message is valid else send the erroneous message to a1p(topic:errorMsgQueue). I have the below application. yml file   </p>
  
  <p>spring:
    cloud:
      stream:
        bindings:
          errorMsgQueue:
            binder: kafka1
            destination: errorMsgQueue
            contentType: application/json
          input:
            binder: kafka2
            content-type: application/x-java-object;type=com.comcast.activation.message.vo.ActivationDataInfo
            destination: test
            group: prac<br>
          activationMsgQueue:
            binder: kafka3
            destination: test
            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:
            type: kafka
            environment:
              spring:
               cloud:
                 stream: 
                   kafka:
                     binder:
                       brokers: caapmsg-as-a3p.sys.comcast.net
                       zk-nodes: caapmsg-as-a3p.sys.comcast.net
          kafka3:
            type: kafka
            environment:
              spring:
               cloud:
                 stream: 
                   kafka:
                     binder:
                       brokers: caapmsg-as-a1p.sys.comcast.net
                       zk-nodes: caapmsg-as-a1p.sys.comcast.net
        default-binder: kafka2          </p>
  
  <p>I am still getting an infinite loop. What am i doing wrong?</p>