Support for copying metric listeners from existing Producers (#784)

* roughin- withProducerMetricsListenerFromExisting

* tests and some refactorings

* Renamed method to withMetricsListenersFromExisting

* Renamed method to withMetricsListenerFromExisting

---------

Co-authored-by: Sunjeet Singh <sunjeets@netflix.com>
Co-authored-by: Jegan <jponrama@netflix.com>

Author: 3 people

hollow/src/main/java/com/netflix/hollow/api/producer/AbstractHollowProducer.java

@@ -30,6 +30,7 @@
 import com.netflix.hollow.api.producer.fs.HollowFilesystemBlobStager;
 import com.netflix.hollow.api.producer.listener.CycleListener;
 import com.netflix.hollow.api.producer.listener.HollowProducerEventListener;
+import com.netflix.hollow.api.producer.metrics.AbstractProducerMetricsListener;
 import com.netflix.hollow.api.producer.validation.ValidationResult;
 import com.netflix.hollow.api.producer.validation.ValidationStatus;
 import com.netflix.hollow.api.producer.validation.ValidationStatusException;
@@ -125,6 +126,11 @@ public AbstractHollowProducer(
                 b.hashCodeFinder, b.doIntegrityCheck, b.updatePlanBlobVerifier);
     }
 
+    private final HollowProducerListener producerMetricsListener;
+    HollowProducerListener getProducerMetricsListener() {
+        return producerMetricsListener;
+    }
+
     private AbstractHollowProducer(
             HollowProducer.BlobStager blobStager,
             HollowProducer.Publisher publisher,
@@ -172,6 +178,14 @@ private AbstractHollowProducer(
         this.blobStorageCleaner = blobStorageCleaner;
 
         this.listeners = new ProducerListenerSupport(eventListeners.stream().distinct().collect(toList()));
+        HollowProducerEventListener temp = null;
+        for (HollowProducerEventListener l : eventListeners) {
+            if (l instanceof AbstractProducerMetricsListener) {
+                temp = l;
+            }
+        }
+        this.producerMetricsListener = (HollowProducerListener) temp;
+
 
         this.metrics = new HollowProducerMetrics();
         this.metricsCollector = metricsCollector;

hollow/src/main/java/com/netflix/hollow/api/producer/HollowProducer.java

@@ -25,6 +25,7 @@
 import com.netflix.hollow.api.producer.enforcer.SingleProducerEnforcer;
 import com.netflix.hollow.api.producer.fs.HollowFilesystemBlobStager;
 import com.netflix.hollow.api.producer.listener.HollowProducerEventListener;
+import com.netflix.hollow.api.producer.metrics.AbstractProducerMetricsListener;
 import com.netflix.hollow.api.producer.validation.ValidatorListener;
 import com.netflix.hollow.core.read.engine.HollowReadStateEngine;
 import com.netflix.hollow.core.schema.HollowSchema;
@@ -111,6 +112,12 @@
  */
 public class HollowProducer extends AbstractHollowProducer {
 
+
+    static HollowProducerAccessor ACCESSOR;
+    static void setAccessor(HollowProducerAccessor accessor) {
+        ACCESSOR = accessor;
+    }
+
     /*
      * HollowProducer and HollowProducer.Incremental extend from a package protected AbstractHollowProducer
      * for sharing common functionality.
@@ -746,6 +753,31 @@ public static class Builder<B extends HollowProducer.Builder<B>> {
         ProducerOptionalBlobPartConfig optionalPartConfig = null;
         HollowConsumer.UpdatePlanBlobVerifier updatePlanBlobVerifier = HollowConsumer.UpdatePlanBlobVerifier.DEFAULT_INSTANCE;
 
+        protected HollowProducerEventListener customProducerMetricsListener = null;
+
+        public B withMetricsListenerFromExisting(HollowProducer producer) {
+            HollowProducerAccessHelper.ensureInit();
+            this.customProducerMetricsListener = HollowProducer.ACCESSOR.getProducerMetricsListener(producer);
+            return (B) this;
+        }
+
+        protected HollowProducerEventListener getCustomProducerMetricsListener() {
+            return customProducerMetricsListener;
+        }
+
+        private void overrideProducerMetricsListenerIfNeeded() {
+            if (customProducerMetricsListener == null) {
+                return;
+            }
+
+            // If the builder already has configured a producer metrics listener, do not override it.
+            boolean hasProducerMetricsListener = eventListeners.stream()
+                    .anyMatch(l -> l instanceof AbstractProducerMetricsListener);
+            if (!hasProducerMetricsListener) {
+                eventListeners.add(customProducerMetricsListener);
+            }
+        }
+
         public B withBlobStager(HollowProducer.BlobStager stager) {
             this.stager = stager;
             return (B) this;
@@ -977,6 +1009,7 @@ protected void checkArguments() {
          */
         public HollowProducer build() {
             checkArguments();
+            overrideProducerMetricsListenerIfNeeded();
             return new HollowProducer(this);
         }
 
@@ -988,6 +1021,7 @@ public HollowProducer build() {
          */
         public HollowProducer.Incremental buildIncremental() {
             checkArguments();
+            overrideProducerMetricsListenerIfNeeded();
             return new HollowProducer.Incremental(this);
         }
     }

hollow/src/main/java/com/netflix/hollow/api/producer/HollowProducerAccessHelper.java

@@ -0,0 +1,15 @@
+package com.netflix.hollow.api.producer;
+
+final class HollowProducerAccessHelper {
+    static {
+        HollowProducer.setAccessor(new HollowProducerAccessor() {
+            @Override
+            public HollowProducerListener getProducerMetricsListener(HollowProducer producer) {
+                return producer.getProducerMetricsListener();
+            }
+        });
+    }
+
+    // force class to load (builder will call this)
+    static void ensureInit() {}
+}

hollow/src/main/java/com/netflix/hollow/api/producer/HollowProducerAccessor.java

@@ -0,0 +1,5 @@
+package com.netflix.hollow.api.producer;
+
+public interface HollowProducerAccessor {
+    HollowProducerListener getProducerMetricsListener(HollowProducer producer);
+}

hollow/src/test/java/com/netflix/hollow/api/producer/HollowProducerBuilderWithListenersTest.java

@@ -0,0 +1,52 @@
+package com.netflix.hollow.api.producer;
+
+import static org.junit.Assert.assertSame;
+
+import com.netflix.hollow.api.producer.HollowProducer.PublishArtifact;
+import com.netflix.hollow.api.producer.metrics.AbstractProducerMetricsListener;
+import java.io.IOException;
+import org.junit.Test;
+
+public class HollowProducerBuilderWithListenersTest {
+
+    private static final class NoopPublisher implements HollowProducer.Publisher {
+        @Override
+        public void publish(PublishArtifact publishArtifact) {}
+    }
+
+    private static final class TestMetricsListener extends AbstractProducerMetricsListener {
+    }
+
+    @Test
+    public void withMetricsListenersFromExisting_reusesMetricsListenerWhenNotExplicitlySet() throws IOException {
+        TestMetricsListener metricsListener = new TestMetricsListener();
+
+        HollowProducer producer1 = HollowProducer.withPublisher(new NoopPublisher())
+                .withListener(metricsListener)
+                .build();
+
+        assertSame(metricsListener, producer1.getProducerMetricsListener());
+
+        HollowProducer producer2 = HollowProducer.withPublisher(new NoopPublisher())
+                .withMetricsListenerFromExisting(producer1)
+                .build();
+
+        assertSame(producer1.getProducerMetricsListener(), producer2.getProducerMetricsListener());
+    }
+
+    @Test
+    public void withMetricsListenersFromExisting_doesNotOverrideProvidedMetricsListener() throws IOException {
+        TestMetricsListener metricsListener = new TestMetricsListener();
+        HollowProducer producer1 = HollowProducer.withPublisher(new NoopPublisher())
+                .withListener(metricsListener)
+                .build();
+
+        TestMetricsListener metricsListener2 = new TestMetricsListener();
+        HollowProducer producer2 = HollowProducer.withPublisher(new NoopPublisher())
+                .withListener(metricsListener2)
+                .withMetricsListenerFromExisting(producer1)
+                .build();
+
+        assertSame(metricsListener2, producer2.getProducerMetricsListener());
+    }
+}