Hi all, I am proposing the following enhancements to the Malhar JMS input operator. Let me know if you have any comments.
Enhancements are proposed to the malhar JMS Input Operator (in the package com.datatorrent.lib.io.jms) to make it usable with any JMS compatible message broker for basic input functionality. For now we will make this input operator work with ActiveMQ and Amazon SQS through the JMS interface. The current code contains implementation and tests for ActiveMQ, and we will add support and tests for SQS without breaking the ActiveMQ support. The motivation for this enhancement is as follows. JMS is NOT a wire protocol but just an API specification for Java programs, so JMS is just a “wrapper” around an actual implementation library that talks to the respective message broker. Moreover each message broker has its own semantics for more advanced features like partitioning or dynamic scaling that are impossible or difficult to capture via the JMS abstraction. With this enhancement, we expect the JMS input operator to be usable as a generic “JMS input operator” with no support for advanced features like partionability or dynamic scalability. Full featured input operators for SQS and ActiveMQ will be developed separately (without necessarily using the JMS interface but the most appropriate interface) and the design and implementation of those operators are out of scope here. The enhancement includes: - adding amazon-sqs-java-messaging-lib and aws-android-sdk-sqs libraries to the pom - adding test cases for SQS, matching those for ActiveMQ. - making both sets of test cases work - cleaning up the existing code (e.g. removing hard coded values like “TEST.FOO” in JMSBase) SQS ConsiderationsNo Support for Topics SQS only supports queues, so topics will not be supported for the SQS variant. Idempotency We will implement a simple idempotent design that uses the WindowDataManager to store the whole message and deletes the message from SQS so we are not impacted by SQS’s visibility timeout feature. Sanjay
