NIFI-6926: Fixed memory leak in NiFiAtlasHook

[turcsanyip]

Thank you for submitting a contribution to Apache NiFi.

Please provide a short description of the PR here:

Description of PR

Enables X functionality; fixes bug NIFI-YYYY.

In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:

For all changes:

• Is there a JIRA ticket associated with this PR? Is it referenced
  in the commit message?

• Does your PR title start with NIFI-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.

• Has your PR been rebased against the latest commit within the target branch (typically master)?

• Is your initial contribution a single, squashed commit? Additional commits in response to PR reviewer feedback should be made on this branch and pushed to allow change tracking. Do not squash or use --force when pushing to allow for clean monitoring of changes.

For code changes:

• Have you ensured that the full suite of tests is executed via mvn -Pcontrib-check clean install at the root nifi folder?
• Have you written or updated unit tests to verify your changes?
• Have you verified that the full build is successful on both JDK 8 and JDK 11?
• If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under ASF 2.0?
• If applicable, have you updated the LICENSE file, including the main LICENSE file under nifi-assembly?
• If applicable, have you updated the NOTICE file, including the main NOTICE file found under nifi-assembly?
• If adding new Properties, have you added .displayName in addition to .name (programmatic access) for each of the new properties?

For documentation related changes:

• Have you ensured that format looks appropriate for the output in which it is rendered?

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.

[belugabehr]

The fact that the send method accepts a function makes me think that the send happens asynchronously. Are you sure that send is a blocking method? If not, clearing the messages will cause a ConcurrentModificationException

[turcsanyip]

Good point, I've checked it now: the items from list are streamed / collected into other data structures at the beginning of send() and the list itself never used later, so it shouldn't be a problem.
Though, just to be on the safe side, I'll change it to create a new list after every commit.

[belugabehr]

Hello @turcsanyip

I hope you will consider another solution:

https://github.com/belugabehr/nifi/blob/6c224dbf08731f6d90366911f3e0aaf24c5010bd/nifi-nar-bundles/nifi-atlas-bundle/nifi-atlas-reporting-task/src/main/java/org/apache/nifi/atlas/hook/NiFiAtlasHook.java

I have implemented this change with a BlockingQueue because I believe adding to the collection and submitting it for processing can occur on different threads... it makes sense that it would: thread from repository is filling it and every N minutes, a new thread sends out the messages. Also allows one to re-use the queue (don't have to replace it every iteration).

[turcsanyip]

@belugabehr Thanks for your suggestion and in a concurrent environment it would be perfect.
In our case both addMessage() and commitMessages() are called from the same place and on the same thread: ReportLineageToAtlas.consumeNiFiProvenanceEvents().
Although we could enhance it to support concurrent access as well, but NiFiAtlasHook explicitly declares itself as not being thread safe in its javadoc. That's why I would not add concurrency handling code to it and mix thread safe / non thread safe behaviours.

[belugabehr]

@turcsanyip Thank you for the review! I actually liked your first solution of using clear() better because I think using immutable values are easier to work with when debugging. Sorry to nit, but could you please consider reverting back to using clear()? Perhaps you can make a copy of the list and pass the copy into the method before calling clear(), but if you're confident that the send method is synchronous, then you don't even need to make a copy of the list.

[belugabehr]

@turcsanyip Also, does the list of messages need to be flushed in the close() method?

[turcsanyip]

@belugabehr Passing a copy and clear the list sounds reasonable, pls. check my last commit.

Regarding flushing (calling commit) in close():
The reporting task stores the id of the last committed message in its persistent state and it happens after the commit returned.

nifi/nifi-nar-bundles/nifi-extension-utils/nifi-reporting-utils/src/main/java/org/apache/nifi/reporting/util/provenance/ProvenanceEventConsumer.java

Lines 218 to 223 in 41fef55

	if (!filteredEvents.isEmpty()) {
	// Executes callback.
	consumer.accept(componentMapHolder, filteredEvents);
	}
	firstEventId = updateLastEventId(rawEvents, stateManager);

So even if close() gets called with uncommitted messages in the list, the next execution of the reporting task will continue where it left off and reloads the uncommitted messages from the provenance repository and process them again. The at-least-once delivery is guaranteed.
If I understand correctly, this was your concern.

[belugabehr]

@turcsanyip Yes. You understand my concern. Thanks for the collaboration and running down all of my concerns. Looks good to me.

[mcgilman]

Thanks @turcsanyip for the PR! Thanks @belugabehr for the review! This has been merged to master.