NIFI-6926: Fixed memory leak in NiFiAtlasHook

NIFI-6926: Use new instance of list instead of clearing it
NIFI-6926: Logging the number of messages to be sent to Atlas.
NIFI-6926: Pass a copy of the messages list to send() and clear the original list.

This closes #3915

nifi-nar-bundles/nifi-atlas-bundle/nifi-atlas-reporting-task/src/main/java/org/apache/nifi/atlas/hook/NiFiAtlasHook.java

@@ -55,9 +55,11 @@ public class NiFiAtlasHook extends AtlasHook implements LineageContext {
     }
 
     public void commitMessages() {
-        final NotificationSender notificationSender = new NotificationSender();
+        final NotificationSender notificationSender = createNotificationSender();
         notificationSender.setAtlasClient(atlasClient);
-        notificationSender.send(messages, this::notifyEntities);
+        List<HookNotificationMessage> messagesBatch = new ArrayList<>(messages);
+        messages.clear();
+        notificationSender.send(messagesBatch, this::notifyEntities);
     }
 
     public void close() {
@@ -65,4 +67,12 @@ public class NiFiAtlasHook extends AtlasHook implements LineageContext {
             notificationInterface.close();
         }
     }
+
+    NotificationSender createNotificationSender() {
+        return new NotificationSender();
+    }
+
+    List<HookNotificationMessage> getMessages() {
+        return messages;
+    }
 }

nifi-nar-bundles/nifi-atlas-bundle/nifi-atlas-reporting-task/src/main/java/org/apache/nifi/atlas/hook/NotificationSender.java

@@ -191,6 +191,8 @@ class NotificationSender {
      * @param notifier responsible for sending notification messages, its accept method can be called multiple times
      */
     void send(final List<HookNotification.HookNotificationMessage> messages, final Consumer<List<HookNotification.HookNotificationMessage>> notifier) {
+        logger.info("Sending {} messages to Atlas", messages.size());
+
         final Metrics metrics = new Metrics();
         try {
             metrics.totalMessages = messages.size();

nifi-nar-bundles/nifi-atlas-bundle/nifi-atlas-reporting-task/src/test/java/org/apache/nifi/atlas/hook/TestNiFiAtlasHook.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.atlas.hook;
+
+import org.apache.atlas.notification.hook.HookNotification;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+
+public class TestNiFiAtlasHook {
+
+    private NiFiAtlasHook hook;
+
+    @Before
+    public void setUp() {
+        hook = new NiFiAtlasHook() {
+            @Override
+            NotificationSender createNotificationSender() {
+                return mock(NotificationSender.class);
+            }
+        };
+    }
+
+    @Test
+    public void messagesListShouldContainMessagesAfterAddMessage() {
+        hook.addMessage(new HookNotification.HookNotificationMessage(HookNotification.HookNotificationType.ENTITY_CREATE, "nifi"));
+        hook.addMessage(new HookNotification.HookNotificationMessage(HookNotification.HookNotificationType.ENTITY_PARTIAL_UPDATE, "nifi"));
+
+        assertEquals(2, hook.getMessages().size());
+    }
+
+    @Test
+    public void messagesListShouldBeCleanedUpAfterCommit() {
+        hook.addMessage(new HookNotification.HookNotificationMessage(HookNotification.HookNotificationType.ENTITY_CREATE, "nifi"));
+        hook.addMessage(new HookNotification.HookNotificationMessage(HookNotification.HookNotificationType.ENTITY_PARTIAL_UPDATE, "nifi"));
+
+        hook.commitMessages();
+
+        assertTrue(hook.getMessages().isEmpty());
+    }
+}