Track and propagate MinRequestTime on the /reaggregate path (#121)

Co-authored-by: William Ehlhardt <[email protected]>

Author: orborde

src/main/java/com/arpnetworking/clusteraggregator/aggregation/StreamingAggregationBucket.java

@@ -26,6 +26,7 @@
 import org.joda.time.DateTime;
 
 import java.util.Map;
+import java.util.Optional;
 
 /**
  * Container class that holds aggregation pending records.
@@ -84,6 +85,10 @@ public boolean isSpecified(final Statistic statistic) {
         return _specified.getOrDefault(statistic, false);
     }
 
+    public Optional<DateTime> getMinRequestTime() {
+        return _minRequestTime;
+    }
+
     /**
      * Add <code>AggregatedData</code> instance.
      *
@@ -122,9 +127,23 @@ public void update(final CombinedMetricData datum) {
                         .log();
             }
         }
+
+        updateMinRequestTime(datum);
+    }
+
+    private void updateMinRequestTime(final CombinedMetricData datum) {
+        if (datum.getMinRequestTime().isPresent()) {
+            if (!_minRequestTime.isPresent()) {
+                _minRequestTime = datum.getMinRequestTime();
+            }
+            if (datum.getMinRequestTime().get().isBefore(_minRequestTime.get())) {
+                _minRequestTime = datum.getMinRequestTime();
+            }
+        }
     }
 
     private final DateTime _periodStart;
+    private Optional<DateTime> _minRequestTime = Optional.empty();
     private final Map<Statistic, Calculator<?>> _data = Maps.newHashMap();
     private final Map<Statistic, Boolean> _specified = Maps.newHashMap();
 

src/main/java/com/arpnetworking/clusteraggregator/aggregation/StreamingAggregator.java

@@ -172,6 +172,7 @@ public Receive createReceive() {
                                         .setConditions(ImmutableList.of())
                                         .setPeriod(_period)
                                         .setStart(bucket.getPeriodStart())
+                                        .setMinRequestTime(bucket.getMinRequestTime().orElse(null))
                                         .build();
                                 _emitter.tell(periodicData, getSelf());
                             } else {