in flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/QueueingConsumer.java [78:96]
private Delivery handle(Delivery delivery) {
if (delivery == POISON || delivery == null && (shutdown != null || cancelled != null)) {
if (delivery == POISON) {
queue.add(POISON);
if (shutdown == null && cancelled == null) {
throw new IllegalStateException(
"POISON in queue, but null shutdown and null cancelled. "
+ "This should never happen, please report as a BUG");
}
}
if (null != shutdown) {
throw Utility.fixStackTrace(shutdown);
}
if (null != cancelled) {
throw Utility.fixStackTrace(cancelled);
}
}
return delivery;
}