Storm High Availability and Failover - Real-time Analytics with Storm and Cassandra - page 103

public void fail(Object msgId) {
if (msgId instanceof Long) {
final long deliveryTag = (Long) msgId;
if (amqpChannel != null) {
try {
if (amqpChannel.isOpen()) {
if (!this.autoAck) {
amqpChannel.basicReject(deliveryTag,
requeueOnFail);
}
} else {
reconnect();
}
} catch (IOException e) {
log.warn("Failed to reject delivery-tag " +
deliveryTag, e);
}
}
} else {
log.warn(String.format("don't know how to reject(%s:
%s)", msgId.getClass().getName(), msgId));
}
}
public void declareOutputFields(OutputFieldsDeclarer
declarer) {
declarer.declare(new Fields("messages"));
}
}
AMQP Maven dependency that will be required to be introduced in the project pom.xml ,
as shown in the following code:
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.2.1</version>
</dependency>