[
https://issues.apache.org/jira/browse/BAHIR-190?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16843415#comment-16843415
]
ASF GitHub Bot commented on BAHIR-190:
--------------------------------------
lresende commented on issue #53: [BAHIR-190] [activemq] Fixed premature exit on
empty queue
URL: https://github.com/apache/bahir-flink/pull/53#issuecomment-493753758
@Krystex ^^^
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> ActiveMQ connector stops on empty queue
> ---------------------------------------
>
> Key: BAHIR-190
> URL: https://issues.apache.org/jira/browse/BAHIR-190
> Project: Bahir
> Issue Type: Bug
> Components: Flink Streaming Connectors
> Affects Versions: Flink-1.0
> Reporter: Stephan Brosinski
> Priority: Critical
> Fix For: Flink-Next
>
>
> I tried the ActiveMQ Flink Connector. Reading from an ActiveMQ queue, it
> seems to connector exits once there are no more messages in the queue. This
> ends the Flink job processing the stream.
> To me it seems, that the while loop inside the run method (AMQSource.java,
> line 222) should not do a return, but a continue if the message is no
> instance of ByteMessage, e.g. null.
> If I'm right, I can create a pull request showing the change.
> To reproduce:
>
> {code:java}
> ActiveMQConnectionFactory connectionFactory = new
> ActiveMQConnectionFactory("xxx", "xxx",
> "tcp://localhost:61616?jms.redeliveryPolicy.maximumRedeliveries=1");
> AMQSourceConfig<String> amqConfig = new
> AMQSourceConfig.AMQSourceConfigBuilder<String>()
> .setConnectionFactory(connectionFactory)
> .setDestinationName("test")
> .setDestinationType(DestinationType.QUEUE)
> .setDeserializationSchema(new SimpleStringSchema())
> .build();
> AMQSource<String> amqSource = new AMQSource<>(amqConfig);
> env.addSource(amqSource).print()
> env.setParallelism(1).execute("ActiveMQ Consumer");{code}
> Then point the Flink job at an empty ActiveMQ queue.
>
> Not sure if this is a bug, but it's not what I expected when I used the
> connector.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
