Step hang after unexpected exception : java.net.SocketException: Connection timed out

[dshvedchenko]

2015/06/18 23:40:04 - RECEIVE.0 - Closing AMQP channel
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod - ERROR (version 5.3.0.0-213, build 1 from 2015-02-02_12-17-08 by buildguy) : EIS_AMQP/landable_visits_prod : Unexpected error at the end of the transformation
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod - ERROR (version 5.3.0.0-213, build 1 from 2015-02-02_12-17-08 by buildguy) : com.rabbitmq.client.ShutdownSignalException: connection error; reason: java.net.SocketException: Connection timed out
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:67)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:37)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:349)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:569)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:501)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:494)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.instaclick.pentaho.plugin.amqp.AMQPPlugin.closeAmqp(AMQPPlugin.java:404)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.instaclick.pentaho.plugin.amqp.AMQPPlugin.flush(AMQPPlugin.java:383)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.instaclick.pentaho.plugin.amqp.AMQPPlugin.access$200(AMQPPlugin.java:35)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.instaclick.pentaho.plugin.amqp.AMQPPlugin$2.transFinished(AMQPPlugin.java:98)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at org.pentaho.di.trans.Trans.fireTransFinishedListeners(Trans.java:1468)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at org.pentaho.di.trans.Trans$1.stepFinished(Trans.java:1227)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at org.pentaho.di.trans.step.BaseStep.markStop(BaseStep.java:2995)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at org.pentaho.di.trans.step.RunThread.run(RunThread.java:121)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at java.lang.Thread.run(Thread.java:745)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod - Caused by: java.net.SocketException: Connection timed out
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at java.net.SocketInputStream.socketRead0(Native Method)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at java.net.SocketInputStream.read(SocketInputStream.java:152)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at java.net.SocketInputStream.read(SocketInputStream.java:122)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at java.io.BufferedInputStream.fill(BufferedInputStream.java:235)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at java.io.BufferedInputStream.read(BufferedInputStream.java:254)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at java.io.DataInputStream.readUnsignedByte(DataInputStream.java:288)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.rabbitmq.client.impl.Frame.readFrom(Frame.java:95)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.rabbitmq.client.impl.SocketFrameHandler.readFrame(SocketFrameHandler.java:131)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod -   at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:533)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod - Transformation detected one or more steps with errors.
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod - Transformation is killing the other steps!

[FabioBatSilva]

It is probably missing a proper error handling here on the WaitingConsumer

[dshvedchenko]

maybe needed in consume()

 try {
        delivery = consumer.nextDelivery(data.waitTimeout);
    } catch (InterruptedException ex) {
        throw new KettleStepException(ex.getMessage(), ex);
    } catch (ShutdownSignalException ex) {
        throw new KettleStepException(ex.getMessage(), ex);
    } catch (ConsumerCancelledException ex) {
        throw new KettleStepException(ex.getMessage(), ex);

--> } finally {
return false;
}

[FabioBatSilva]

Not sure,
if i remember correctly the KettleStepException it's supposed to abort the transformation

[dshvedchenko]

Yes, and then we found control in transListener and invoke flush

com.instaclick.pentaho.plugin.amqp.AMQPPlugin.closeAmqp(AMQPPlugin.java:404)
2015/06/18 23:55:36 - EIS_AMQP/landable_visits_prod - at com.instaclick.pentaho.plugin.amqp.AMQPPlugin.flush(AMQPPlugin.java:383)

code above has trouble with detecting channel or consumer already closed.