merrimanr opened a new pull request #1330: METRON-1968: Messages are lost when a parser produces multiple messages and batch size is greater than 1 URL: https://github.com/apache/metron/pull/1330 ## Contributor Comments This PR represents a fairly significant shift in the Writer class architecture. Currently these classes do not support tuples that result in multiple messages, mainly due to a limitation in the `BulkMessageWriter` interface. The `write` method accepts separate lists of tuples and messages so there is no way to know which tuples are associated with which message. It has worked so far with parsers that only emit a single message from a tuple because a 1 to 1 relationship is assumed in these classes. I experimented with several different approaches to fixing this and tried to follow a strategy that would avoid having to significantly rewrite multiple classes (this was unavoidable in certain places). I change the `BulkMessageWriter.write` interface from: ``` BulkWriterResponse write(String sensorType, WriterConfiguration configurations, Iterable<Tuple> tuples, List<MESSAGE_T> messages) throws Exception; ``` to: ``` BulkWriterResponse write(String sensorType, WriterConfiguration configurations, Map<String, MESSAGE_T> messages) throws Exception; ``` The messages being passed in are now represented as a map where the keys are message ids and the values are the messages. This made changes to the `BulkMessageWriter` implementations easier because they continue to assume a 1 to 1 relationship between message ids (formerly tuples) and messages. Now the writers report which message ids succeeded/failed instead of tuples. The `BulkWriterComponent` continues track batch sizes and timeouts but no longer manages tuples or error handling. Instead it calls `BulkWriterResponseHandler.handleFlush` when messages are flushed. This interface is injected into `BulkWriterComponent` and allows us to properly commit messages according to the requirements of the platform the classes are running on. For now a Storm implementation of `BulkWriterResponseHandler` is setup and passed into the `BulkWriterComponent`. Managing tuple to message relationships as well as tuple acking and error handling are now done in a single class. The bolts now have to setup a `StormBulkWriterResponseHandler` object and add tuples/messages to it. ### Changes Included - The `BulkMessageWriter` is updated and all implementations have been adjusted to conform to the new interfaces. Changes to the implementations are small and straightforward. - Tuple acking and error handling is moved out of `BulkWriterComponent` and into `StormBulkWriterResponseHandler`. - Bolts have been updated with the new `StormBulkWriterResponseHandler` pattern - I noticed `HBaseWriter` and `PcapWriter are no longer being used. Rather than go to the trouble of updating their tests I removed them. - Removed `WriterHandler.handleAck` since acks are no longer handled here. There were also several significant changes that needed to be made to the tests: - The writer tests are updated to match the new `BulkMessageWriter.write` interface. In most cases these changes were simple however I noticed there were no unit tests for `KafkaWriter.write` so I added them. - The parser integration tests now verify the all tuples were acked. - There is now a parser integration test that simulates the use case described in this Jira. The `jsonMapQuery` parser integration test now produces multiple messages from a single tuple and sets the batch size to 5. - A test was added for `StormBulkWriterResponseHandler` that also simulates the use case described in this Jira. Error handling tests that were originally in `BulkWriterComponentTest` were migrated here. ### Testing This has been tested in full dev both for regression and for the use case described in this Jira. I am in the process of writing comprehensive testing instructions and will add them soon. ### Feedback Requested Outside of the standard code review and function testing, they are a few areas that are not 100% clear and I would like feedback on: - Is the fundamental architectural approach solid? Are there any holes I'm not thinking of? - The message ids generated by the bolts and used to track messages in the writer classes are hashes of the messages. Is this good enough? - There is the potential for messages to pile up in `StormBulkWriterResponseHandler` (and probably other `BulkWriterComponent` too) when tuples time out and are replayed. I think this is also an issue now but should we explore some kind of cache that can evict messages after timeouts? - I have added javadocs and am planning on taking another pass and adding more. Are there areas you feel need more/better explanation? My goal is for these classes to be easy to understand and navigate. Is the interaction between the bolts and response handler clear? ## Pull Request Checklist Thank you for submitting a contribution to Apache Metron. Please refer to our [Development Guidelines](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=61332235) for the complete guide to follow for contributions. Please refer also to our [Build Verification Guidelines](https://cwiki.apache.org/confluence/display/METRON/Verifying+Builds?show-miniview) for complete smoke testing guides. In order to streamline the review of the contribution we ask you follow these guidelines and ask you to double check the following: ### For all changes: - [x] Is there a JIRA ticket associated with this PR? If not one needs to be created at [Metron Jira](https://issues.apache.org/jira/browse/METRON/?selectedTab=com.atlassian.jira.jira-projects-plugin:summary-panel). - [x] Does your PR title start with METRON-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character. - [x] Has your PR been rebased against the latest commit within the target branch (typically master)? ### For code changes: - [ ] Have you included steps to reproduce the behavior or problem that is being changed or addressed? - [ ] Have you included steps or a guide to how the change may be verified and tested manually? - [x] Have you ensured that the full suite of tests and checks have been executed in the root metron folder via: ``` mvn -q clean integration-test install && dev-utilities/build-utils/verify_licenses.sh ``` - [x] Have you written or updated unit tests and or integration tests to verify your changes? - [x] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)? - [x] Have you verified the basic functionality of the build by building and running locally with Vagrant full-dev environment or the equivalent? ### For documentation related changes: - [x] Have you ensured that format looks appropriate for the output in which it is rendered by building and verifying the site-book? If not then run the following commands and the verify changes via `site-book/target/site/index.html`: ``` cd site-book mvn site ``` #### Note: Please ensure that once the PR is submitted, you check travis-ci for build issues and submit an update to your PR as soon as possible. It is also recommended that [travis-ci](https://travis-ci.org) is set up for your personal repository such that your branches are built there before submitting a pull request.
---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
