ATLAS-3878: Notifications: Improve Memory Usage for HBase Audits Writing

intg/src/main/java/org/apache/atlas/AtlasConfiguration.java

@@ -47,6 +47,7 @@ public enum AtlasConfiguration {
     NOTIFICATION_MESSAGE_COMPRESSION_ENABLED("atlas.notification.message.compression.enabled", true),
     NOTIFICATION_SPLIT_MESSAGE_SEGMENTS_WAIT_TIME_SECONDS("atlas.notification.split.message.segments.wait.time.seconds", 15 * 60),
     NOTIFICATION_SPLIT_MESSAGE_BUFFER_PURGE_INTERVAL_SECONDS("atlas.notification.split.message.buffer.purge.interval.seconds", 5 * 60),
+    NOTIFICATION_FIXED_BUFFER_ITEMS_INCREMENT_COUNT("atlas.notification.fixed.buffer.items.increment.count", 10),
 
     NOTIFICATION_CREATE_SHELL_ENTITY_FOR_NON_EXISTING_REF("atlas.notification.consumer.create.shell.entity.for.non-existing.ref", true),
     REST_API_CREATE_SHELL_ENTITY_FOR_NON_EXISTING_REF("atlas.rest.create.shell.entity.for.non-existing.ref", false),

intg/src/main/java/org/apache/atlas/model/Clearable.java

+/**
+ * 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.atlas.model;
+
+public interface Clearable {
+    void clear();
+}

intg/src/main/java/org/apache/atlas/model/audit/EntityAuditEventV2.java

@@ -21,6 +21,7 @@ import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import org.apache.atlas.model.Clearable;
 import org.apache.atlas.model.instance.AtlasEntity;
 import org.apache.atlas.model.instance.AtlasEntityHeader;
 import org.apache.atlas.type.AtlasType;
@@ -44,7 +45,7 @@ import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditType.EN
 @JsonIgnoreProperties(ignoreUnknown=true)
 @XmlRootElement
 @XmlAccessorType(XmlAccessType.PROPERTY)
-public class EntityAuditEventV2 implements Serializable {
+public class EntityAuditEventV2 implements Serializable, Clearable {
     public enum EntityAuditType { ENTITY_AUDIT_V1, ENTITY_AUDIT_V2 }
 
     public enum EntityAuditActionV2 {
@@ -255,6 +256,19 @@ public class EntityAuditEventV2 implements Serializable {
         return ret;
     }
 
+    @JsonIgnore
+    @Override
+    public void clear() {
+        entityId = null;
+        timestamp = 0L;
+        user = null;
+        action = null;
+        details = null;
+        eventKey = null;
+        entity = null;
+        type = null;
+    }
+
     private String getJsonPartFromDetails() {
         String ret = null;
         if(StringUtils.isNotEmpty(details)) {

intg/src/main/java/org/apache/atlas/utils/FixedBufferList.java

+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.atlas.utils;
+
+import org.apache.atlas.model.Clearable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class FixedBufferList<T extends Clearable> {
+    private static final Logger LOG = LoggerFactory.getLogger(FixedBufferList.class);
+
+    private final Class<T> itemClass;
+    private final ArrayList<T> buffer;
+    private final int incrementCapacityBy;
+
+    private int length;
+
+    public FixedBufferList(Class<T> clazz) {
+        this(clazz, 1);
+    }
+
+    public FixedBufferList(Class<T> clazz, int incrementCapacityBy) {
+        this.incrementCapacityBy = (incrementCapacityBy <= 0 ? 1 : incrementCapacityBy);
+        this.itemClass = clazz;
+        this.buffer = new ArrayList<>();
+    }
+
+    public T next() {
+        request(length + 1);
+        return buffer.get(length++);
+    }
+
+    public List<T> toList() {
+        return this.buffer.subList(0, this.length);
+    }
+
+    public void reset() {
+        for (int i = 0; i < buffer.size(); i++) {
+            buffer.get(i).clear();
+        }
+
+        this.length = 0;
+    }
+
+    private void request(int requestedCapacity) {
+        if (requestedCapacity <= this.buffer.size()) {
+            return;
+        }
+
+        int oldCapacity = this.buffer.size();
+        int newCapacity = oldCapacity + this.incrementCapacityBy;
+        this.buffer.ensureCapacity(newCapacity);
+        instantiateItems(oldCapacity, newCapacity);
+
+        LOG.info("FixedBufferList: Requested: {} From: {} To:{}", requestedCapacity, oldCapacity, newCapacity);
+    }
+
+    private void instantiateItems(int startIndex, int maxSize) {
+        for (int i = startIndex; i < maxSize; i++) {
+            try {
+                this.buffer.add(itemClass.newInstance());
+            } catch (InstantiationException e) {
+                LOG.error("FixedBufferList: InstantiationException: Instantiation failed!", e);
+            } catch (IllegalAccessException e) {
+                LOG.error("FixedBufferList: IllegalAccessException: Instantiation failed!", e);
+            }
+        }
+    }
+}

intg/src/test/java/org/apache/atlas/utils/FixedBufferListTest.java

+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.atlas.utils;
+
+import org.apache.atlas.model.Clearable;
+import org.apache.commons.lang.StringUtils;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+
+public class FixedBufferListTest {
+    private String STR_PREFIX = "str:%s";
+
+    public static class Spying implements Clearable {
+        public static AtomicInteger callsToCtor = new AtomicInteger();
+        public static AtomicInteger callsToClear = new AtomicInteger();
+
+        private int anInt;
+        private String aString;
+        private long aLong;
+
+        public Spying() {
+            callsToCtor.incrementAndGet();
+        }
+
+        @Override
+        public void clear() {
+            callsToClear.incrementAndGet();
+
+            anInt = 0;
+            aString = StringUtils.EMPTY;
+            aLong = 0;
+        }
+
+        public static void resetCounters() {
+            Spying.callsToCtor.set(0);
+            Spying.callsToClear.set(0);
+        }
+    }
+
+    private static class SpyingFixedBufferList extends FixedBufferList<Spying> {
+        public SpyingFixedBufferList(int incrementCapacityFactor) {
+            super(Spying.class, incrementCapacityFactor);
+        }
+    }
+
+    @Test
+    public void instantiateListWithParameterizedClass() {
+        FixedBufferList<Spying> list = new FixedBufferList<>(Spying.class);
+        assertNotNull(list);
+    }
+
+    @Test
+    public void createdBasedOnInitialSize() {
+        Spying.resetCounters();
+
+        int incrementByFactor = 2;
+        SpyingFixedBufferList fixedBufferList = new SpyingFixedBufferList(incrementByFactor);
+        addElements(fixedBufferList, 0, 3);
+
+        List<Spying> list = fixedBufferList.toList();
+        assertSpyingList(list, 3);
+        assertEquals(Spying.callsToCtor.get(), incrementByFactor * 2);
+    }
+
+    @Test (dependsOnMethods = "createdBasedOnInitialSize")
+    public void bufferIncreasesIfNeeded() {
+        Spying.resetCounters();
+
+        int incrementSizeBy = 5;
+        SpyingFixedBufferList fixedBufferList = new SpyingFixedBufferList(incrementSizeBy);
+        addElements(fixedBufferList, 0, incrementSizeBy);
+        List<Spying> spyings = fixedBufferList.toList();
+        assertEquals(spyings.size(), incrementSizeBy);
+        assertEquals(Spying.callsToCtor.get(), incrementSizeBy);
+
+        fixedBufferList.reset();
+        addElements(fixedBufferList, 0, incrementSizeBy * 2);
+        spyings = fixedBufferList.toList();
+        assertEquals(Spying.callsToCtor.get(), incrementSizeBy * 2);
+        assertSpyingList(spyings, incrementSizeBy * 2);
+        assertEquals(Spying.callsToClear.get(), incrementSizeBy);
+    }
+
+    @Test
+    public void retrieveEmptyList() {
+        int size = 5;
+        SpyingFixedBufferList fixedBufferList = new SpyingFixedBufferList(size);
+
+        List<Spying> list = fixedBufferList.toList();
+        assertEquals(list.size(), 0);
+
+        addElements(fixedBufferList, 0, 3);
+        list = fixedBufferList.toList();
+        assertEquals(list.size(), 3);
+    }
+
+    private void assertSpyingList(List<Spying> list, int expectedSize) {
+        assertEquals(list.size(), expectedSize);
+        for (int i1 = 0; i1 < list.size(); i1++) {
+            Assert.assertNotNull(list.get(i1));
+            assertSpying(list.get(i1), i1);
+        }
+    }
+
+    private void assertSpying(Spying spying, int i) {
+        assertEquals(spying.aLong, i);
+        assertEquals(spying.anInt, i);
+        assertEquals(spying.aString, String.format(STR_PREFIX, i));
+    }
+
+    private Spying createSpyingClass(Spying spying, int i) {
+        spying.aLong = i;
+        spying.anInt = i;
+        spying.aString = String.format(STR_PREFIX, i);
+
+        return spying;
+    }
+
+    private void addElements(SpyingFixedBufferList fixedBufferList, int startIndex, int numElements) {
+        for (int i = startIndex; i < (startIndex + numElements); i++) {
+            Spying spyForUpdate = fixedBufferList.next();
+            createSpyingClass(spyForUpdate, i);
+        }
+    }
+}