[FLINK-21040] The exception of "Unknown consumerTag" may throw when RMQSource close() is called.

• Type: Bug
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Affects Version/s: 1.6.3, 1.6.4, 1.7.2, 1.8.0, 1.8.1, 1.8.2, 1.8.3, 1.9.0, 1.9.1, 1.9.2, 1.9.3, 1.10.0, 1.10.1, 1.10.2, 1.11.0, 1.11.1, 1.11.2, 1.11.3, 1.12.0, 1.12.1
• Component/s: Connectors/ RabbitMQ
• Assignee: Unassigned
• Reporter: pcalpha
• Created: 20/Jan/21 02:10
• Updated: 20/Jan/21 02:10

Description

The log is below

2020-10-27 21:00:57  [ Source: rabbitmqSource -> Flat Map (1/1):292311427 ] - [ ERROR ]  Error during disposal of stream operator.2020-10-27 21:00:57  [ Source: rabbitmqSource -> Flat Map (1/1):292311427 ] - [ ERROR ]  Error during disposal of stream operator.java.lang.RuntimeException: Error while cancelling RMQ consumer on xxxxxxxQueueName at 10.xx.xx.102 at org.apache.flink.streaming.connectors.rabbitmq.RMQSource.close(RMQSource.java:196) at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:43) at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.dispose(AbstractUdfStreamOperator.java:117) at org.apache.flink.streaming.runtime.tasks.StreamTask.disposeAllOperators(StreamTask.java:703) at org.apache.flink.streaming.runtime.tasks.StreamTask.cleanUpInvoke(StreamTask.java:635) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:542) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546) at java.base/java.lang.Thread.run(Thread.java:834)Caused by: java.io.IOException: Unknown consumerTag at com.rabbitmq.client.impl.ChannelN.basicCancel(ChannelN.java:1285) at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.basicCancel(AutorecoveringChannel.java:482) at org.apache.flink.streaming.connectors.rabbitmq.RMQSource.close(RMQSource.java:192) ... 8 more