[FLINK-12093] Apache Flink:Active MQ consumer job is getting finished after first message consume.

• Type: Bug
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.7.2
• Component/s: API / DataStream
• Labels: auto-deprioritized-major auto-deprioritized-minor

Environment

Working in my local IDE(Eclipse).

• Assignee: Unassigned
• Reporter: shiv kumar
• Created: 03/Apr/19 07:23
• Updated: 26/Nov/21 22:38

Description

Hi Team,

 

Below is my the code the the execution environment to run the Apache Flink job that's consume message from ActiveMQ topic::

 

StreamExecutionEnvironment env = createExecutionEnvironment();

connectionFactory = new ActiveMQConnectionFactory("*****", "*****.",
"failover:(tcp://amq-master-01:61668)?timeout=3000");

LOG.info("exceptionListener----{}", new AMQExceptionListLocal(LOG, true));

RunningChecker runningChecker = new RunningChecker();
runningChecker.setIsRunning(true);

AMQSourceConfig<String> config = new AMQSourceConfig.AMQSourceConfigBuilder<String>()
.setConnectionFactory(connectionFactory).setDestinationName("test_flink")
.setDeserializationSchema(deserializationSchema).setRunningChecker(runningChecker)
.setDestinationType(DestinationType.TOPIC).build();

amqSource = new AMQSourceLocal<>(config);

LOG.info("Check whether ctx is null ::;;;;{}", amqSource);

DataStream<String> dataMessage = env.addSource(amqSource);

dataMessage.writeAsText("C:/Users/shivkumar/Desktop/flinksjar/output.txt", WriteMode.OVERWRITE);
System.out.println("Step 1");

env.execute("Check ACTIVE_MQ");

 

When we are starting the job, Topic is getting created and message is getting dequeued from that topic.

But After that is getting finished. What Can be done to keep the job running?