Add metric for last announced version in producer

gradle.properties

@@ -0,0 +1 @@
+org.gradle.parallel=true

hollow/src/main/java/com/netflix/hollow/api/metrics/HollowProducerMetrics.java

@@ -23,6 +23,8 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 public class HollowProducerMetrics extends HollowMetrics {
+    private long lastAnnouncedVersion = 0;
+
     private int cyclesCompleted = 0;
     private int cyclesSucceeded = 0;
     private int cycleFailed = 0;
@@ -65,7 +67,7 @@ public void updateCycleMetrics(Status status, HollowProducer.ReadState readState
 
         if(readState != null) {
             HollowReadStateEngine hollowReadStateEngine = readState.getStateEngine();
-            super.update(hollowReadStateEngine, version);
+            update(hollowReadStateEngine, version);
         } else {
             super.update(version);
         }
@@ -138,4 +140,13 @@ public int getReverseDeltasCompleted() {
     public int getReverseDeltasFailed() {
         return reverseDeltasFailed;
     }
+
+    public long getLastAnnouncedVersion() {
+        return lastAnnouncedVersion;
+    }
+
+    protected void update(HollowReadStateEngine hollowReadStateEngine, long version) {
+        lastAnnouncedVersion = version;
+        super.update(hollowReadStateEngine, version);
+    }
 }

hollow/src/test/java/com/netflix/hollow/api/metrics/HollowProducerMetricsTests.java

@@ -37,19 +37,39 @@ public void metricsDoNotBreakWithNullStateEngineInFail() {
         Assert.assertEquals(hollowProducerMetrics.getCycleFailed(), 1);
     }
 
+    @Test
+    public void metricsWhenNothingToPublish() {
+        HollowProducer producer = HollowProducer.withPublisher(blobStore)
+            .withBlobStager(new HollowInMemoryBlobStager())
+            .build();
+
+        producer.runCycle(new HollowProducer.Populator() {
+            public void populate(HollowProducer.WriteState state) throws Exception {
+            }
+        });
+
+        HollowProducerMetrics hollowProducerMetrics = producer.getMetrics();
+        Assert.assertEquals(hollowProducerMetrics.getLastAnnouncedVersion(), 0);
+        Assert.assertEquals(hollowProducerMetrics.getCyclesSucceeded(), 1);
+        Assert.assertEquals(hollowProducerMetrics.getCyclesCompleted(), 1);
+        Assert.assertEquals(hollowProducerMetrics.getTotalPopulatedOrdinals(), 0);
+        Assert.assertEquals(hollowProducerMetrics.getSnapshotsCompleted(), 0);
+    }
+
     @Test
     public void metricsWhenPublishingSnapshot() {
         HollowProducer producer = HollowProducer.withPublisher(blobStore)
                 .withBlobStager(new HollowInMemoryBlobStager())
                 .build();
 
-        producer.runCycle(new HollowProducer.Populator() {
+        long version = producer.runCycle(new HollowProducer.Populator() {
             public void populate(HollowProducer.WriteState state) throws Exception {
                 state.add(Integer.valueOf(1));
             }
         });
 
         HollowProducerMetrics hollowProducerMetrics = producer.getMetrics();
+        Assert.assertEquals(hollowProducerMetrics.getLastAnnouncedVersion(), version);
         Assert.assertEquals(hollowProducerMetrics.getCyclesSucceeded(), 1);
         Assert.assertEquals(hollowProducerMetrics.getCyclesCompleted(), 1);
         Assert.assertEquals(hollowProducerMetrics.getTotalPopulatedOrdinals(), 1);