update akka 2.5.32, use built-in shard passivation instead of rolling our own (#139)

This reverts most of #136, with the exception of the configuration plumbing.

akka 2.5 already has a built-in mechanism for automated actor passivation. It will also handle buffering messages to the actor in case it is resurrected (e.g. if the timeout fires but it is given another message).

This required updating akka to the latest patch release (2.5.16 -> 2.5.32). As a consequence a few deprecated calls were updated, notably PatternsCS -> Patterns.

Author: cwbriones

pom.xml

@@ -101,7 +101,7 @@
 
   <properties>
     <!--Dependency versions-->
-    <akka.version>2.5.16</akka.version>
+    <akka.version>2.5.32</akka.version>
     <akka.http.version>10.1.5</akka.http.version>
     <akka.persistence.cassandra.version>0.89</akka.persistence.cassandra.version>
     <akka.kryo.version>0.5.2</akka.kryo.version>

src/main/java/com/arpnetworking/clusteraggregator/GuiceModule.java

@@ -262,8 +262,7 @@ private java.util.concurrent.CompletionStage<akka.http.javadsl.ServerBinding> pr
         final akka.stream.javadsl.Source<IncomingConnection, CompletionStage<ServerBinding>> binding = http.bind(
                 ConnectHttp.toHost(
                         _configuration.getHttpHost(),
-                        _configuration.getHttpPort()),
-                materializer);
+                        _configuration.getHttpPort()));
         return binding.to(
                 akka.stream.javadsl.Sink.foreach(
                         connection -> connection.handleWithAsyncHandler(routes, materializer)))
@@ -285,13 +284,20 @@ private ActorRef providePeriodicStatsActor(final ActorSystem system, final Metri
     private ActorRef provideAggregatorShardRegion(
             final ActorSystem system,
             final Injector injector,
-            final AggMessageExtractor extractor) {
+            final AggMessageExtractor extractor,
+            @Named("aggregator-liveliness-timeout") final Duration livelinessTimeout) {
         final ClusterSharding clusterSharding = ClusterSharding.get(system);
         final RebalanceConfiguration rebalanceConfiguration = _configuration.getRebalanceConfiguration();
+
+        final ClusterShardingSettings settings =
+                ClusterShardingSettings
+                        .create(system)
+                        .withPassivateIdleAfter(livelinessTimeout);
+
         return clusterSharding.start(
                 "Aggregator",
                 GuiceActorCreator.props(injector, AggregationRouter.class),
-                ClusterShardingSettings.create(system),
+                settings,
                 extractor,
                 new ParallelLeastShardAllocationStrategy(
                         rebalanceConfiguration.getMaxParallel(),
@@ -365,7 +371,6 @@ private Duration provideLivelinessTimeout(final ClusterAggregatorConfiguration c
         return config.getAggregatorLivelinessTimeout();
     }
 
-    @Provides
     @Named("circonus-partition-set")
     @SuppressFBWarnings("UPM_UNCALLED_PRIVATE_METHOD") // Invoked reflectively by Guice
     private PartitionSet provideDatabasePartitionSet(final Injector injector) {

src/main/java/com/arpnetworking/clusteraggregator/Status.java

@@ -21,7 +21,7 @@
 import akka.actor.Props;
 import akka.cluster.Cluster;
 import akka.cluster.MemberStatus;
-import akka.pattern.PatternsCS;
+import akka.pattern.Patterns;
 import akka.remote.AssociationErrorEvent;
 import com.arpnetworking.clusteraggregator.models.MetricsRequest;
 import com.arpnetworking.clusteraggregator.models.PeriodMetrics;
@@ -106,7 +106,7 @@ public Receive createReceive() {
     private void processStatusRequest() {
         // Call the bookkeeper
         final CompletableFuture<ClusterStatusCache.StatusResponse> clusterStateFuture =
-                PatternsCS.ask(
+                Patterns.ask(
                         _clusterStatusCache,
                         new ClusterStatusCache.GetRequest(),
                         Duration.ofSeconds(3))
@@ -115,15 +115,15 @@ private void processStatusRequest() {
                 .toCompletableFuture();
 
         final CompletableFuture<Map<Duration, PeriodMetrics>> localMetricsFuture =
-                PatternsCS.ask(
+                Patterns.ask(
                         _localMetrics,
                         new MetricsRequest(),
                         Duration.ofSeconds(3))
                 .<Map<Duration, PeriodMetrics>>thenApply(new CastMapper<>())
                 .exceptionally(new AsNullRecovery<>())
                 .toCompletableFuture();
 
-        PatternsCS.pipe(
+        Patterns.pipe(
                 CompletableFuture.allOf(
                         clusterStateFuture.toCompletableFuture(),
                         localMetricsFuture.toCompletableFuture())

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

@@ -49,7 +49,6 @@ public class AggregationRouter extends AbstractActor {
      * @param reaggregationDimensions The dimensions to reaggregate over.
      * @param injectClusterAsHost Whether to inject a host dimension based on cluster.
      * @param aggregatorTimeout The time to wait from the start of the period for all data.
-     * @param livelinessTimeout How often to check that a child is still receiving data.
      * @return A new {@link Props}.
      */
     public static Props props(
@@ -58,17 +57,15 @@ public static Props props(
             final String clusterHostSuffix,
             final ImmutableSet<String> reaggregationDimensions,
             final boolean injectClusterAsHost,
-            final Duration aggregatorTimeout,
-            final Duration livelinessTimeout) {
+            final Duration aggregatorTimeout) {
         return Props.create(
                 AggregationRouter.class,
                 metricsListener,
                 emitter,
                 clusterHostSuffix,
                 reaggregationDimensions,
                 injectClusterAsHost,
-                aggregatorTimeout,
-                livelinessTimeout);
+                aggregatorTimeout);
     }
 
     /**
@@ -80,7 +77,6 @@ public static Props props(
      * @param reaggregationDimensions The dimensions to reaggregate over.
      * @param injectClusterAsHost Whether to inject a host dimension based on cluster.
      * @param aggregatorTimeout The time to wait from the start of the period for all data.
-     * @param livelinessTimeout How often to check that a child is still receiving data.
      */
     @Inject
     public AggregationRouter(
@@ -89,18 +85,15 @@ public AggregationRouter(
             @Named("cluster-host-suffix") final String clusterHostSuffix,
             @Named("reaggregation-dimensions") final ImmutableSet<String> reaggregationDimensions,
             @Named("reaggregation-cluster-as-host") final boolean injectClusterAsHost,
-            @Named("reaggregation-timeout") final Duration aggregatorTimeout,
-            @Named("aggregator-liveliness-timeout") final Duration livelinessTimeout) {
-            _streamingChild = context().actorOf(
+            @Named("reaggregation-timeout") final Duration aggregatorTimeout) {
+        _streamingChild = context().actorOf(
                 StreamingAggregator.props(
                         periodicStatistics,
                         emitter,
                         clusterHostSuffix,
                         reaggregationDimensions,
                         injectClusterAsHost,
-                        aggregatorTimeout,
-                        livelinessTimeout
-                ),
+                        aggregatorTimeout),
                 "streaming");
         context().setReceiveTimeout(FiniteDuration.apply(30, TimeUnit.MINUTES));
     }

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

@@ -68,7 +68,6 @@ public class StreamingAggregator extends AbstractActorWithTimers {
      * @param reaggregationDimensions The dimensions to reaggregate over.
      * @param injectClusterAsHost Whether to inject a host dimension based on cluster.
      * @param aggregatorTimeout The time to wait from the start of the period for all data.
-     * @param livelinessTimeout How often to check that this actor is still receiving data.
      * @return A new {@link Props}.
      */
     public static Props props(
@@ -77,17 +76,15 @@ public static Props props(
             final String clusterHostSuffix,
             final ImmutableSet<String> reaggregationDimensions,
             final boolean injectClusterAsHost,
-            final Duration aggregatorTimeout,
-            final Duration livelinessTimeout) {
+            final Duration aggregatorTimeout) {
         return Props.create(
                 StreamingAggregator.class,
                 metricsListener,
                 emitter,
                 clusterHostSuffix,
                 reaggregationDimensions,
                 injectClusterAsHost,
-                aggregatorTimeout,
-                livelinessTimeout);
+                aggregatorTimeout);
     }
 
     /**
@@ -99,7 +96,6 @@ public static Props props(
      * @param reaggregationDimensions The dimensions to reaggregate over.
      * @param injectClusterAsHost Whether to inject a host dimension based on cluster.
      * @param aggregatorTimeout The time to wait from the start of the period for all data.
-     * @param livelinessTimeout How often to check that this actor is still receiving data.
      */
     @Inject
     public StreamingAggregator(
@@ -108,8 +104,7 @@ public StreamingAggregator(
             @Named("cluster-host-suffix") final String clusterHostSuffix,
             @Named("reaggregation-dimensions") final ImmutableSet<String> reaggregationDimensions,
             @Named("reaggregation-cluster-as-host") final boolean injectClusterAsHost,
-            @Named("reaggregation-timeout") final Duration aggregatorTimeout,
-            @Named("aggregator-liveliness-timeout") final Duration livelinessTimeout) {
+            @Named("reaggregation-timeout") final Duration aggregatorTimeout) {
         _periodicStatistics = periodicStatistics;
         _clusterHostSuffix = clusterHostSuffix;
         _reaggregationDimensions = reaggregationDimensions;
@@ -118,7 +113,6 @@ public StreamingAggregator(
         context().setReceiveTimeout(FiniteDuration.apply(30, TimeUnit.MINUTES));
 
         timers().startPeriodicTimer(BUCKET_CHECK_TIMER_KEY, BucketCheck.getInstance(), FiniteDuration.apply(5, TimeUnit.SECONDS));
-        timers().startPeriodicTimer(LIVELINESS_CHECK_TIMER, LIVELINESS_CHECK_MSG, livelinessTimeout);
 
         _emitter = emitter;
     }
@@ -127,8 +121,6 @@ public StreamingAggregator(
     public Receive createReceive() {
         return receiveBuilder()
                 .match(Messages.StatisticSetRecord.class, record -> {
-                    // Mark this actor as live since we're still receiving data.
-                    _live = true;
                     LOGGER.debug()
                             .setMessage("Processing a StatisticSetRecord")
                             .addData("workItem", record)
@@ -190,25 +182,10 @@ public Receive createReceive() {
                         }
                     }
                 })
-                .matchEquals(LIVELINESS_CHECK_MSG, msg -> {
-                    // If we've received data since our last check, reset and wait until another round.
-                    // otherwise shutdown.
-                    if (_live) {
-                        LOGGER.debug()
-                                .setMessage("aggregator is still live, continuing.")
-                                .addContext("actor", self())
-                                .log();
-                        _live = false;
-                        return;
-                    }
-                    LOGGER.debug()
-                            .setMessage("aggregator is stale, requesting shutdown.")
-                            .addContext("actor", self())
-                            .log();
-                    requestShutdownFromParent();
-                })
                 .match(ShutdownAggregator.class, message -> context().stop(self()))
-                .match(ReceiveTimeout.class, message -> requestShutdownFromParent())
+                .match(ReceiveTimeout.class, message -> {
+                    getContext().parent().tell(new ShardRegion.Passivate(ShutdownAggregator.getInstance()), getSelf());
+                })
                 .build();
     }
 
@@ -332,10 +309,6 @@ private void initialize(final Messages.StatisticSetRecord data, final CombinedMe
         }
     }
 
-    private void requestShutdownFromParent() {
-        getContext().parent().tell(new ShardRegion.Passivate(ShutdownAggregator.getInstance()), getSelf());
-    }
-
     private ImmutableMap<String, String> dimensionsToMap(final Messages.StatisticSetRecord statisticSetRecord) {
         // Build a map of dimension key-value pairs dropping any that are to be reaggregated over
         final ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
@@ -383,9 +356,6 @@ private String createHost() {
     private final Set<Statistic> _statistics = Sets.newHashSet();
     private final Duration _aggregatorTimeout;
     private boolean _initialized = false;
-    // This actor is _live if it's received data since the last LIVELINESS_CHECK_MSG.
-    // If this is ever false during a check, the actor will shutdown.
-    private boolean _live = false;
     private Duration _period;
     private String _cluster;
     private String _metric;
@@ -397,9 +367,6 @@ private String createHost() {
                     || entry.getKey().equals(CombinedMetricData.HOST_KEY)
                     || entry.getKey().equals(CombinedMetricData.SERVICE_KEY));
     private static final String BUCKET_CHECK_TIMER_KEY = "bucketcheck";
-    private static final String LIVELINESS_CHECK_TIMER = "livelinesscheck";
-
-    private static final String LIVELINESS_CHECK_MSG = "LIVELINESS_CHECK_MSG";
 
     private static final Logger LOGGER = LoggerFactory.getLogger(StreamingAggregator.class);
     private static final Logger WORK_TOO_OLD_LOGGER = LoggerFactory.getRateLimitLogger(

src/main/java/com/arpnetworking/clusteraggregator/client/HttpSourceActor.java

@@ -86,7 +86,7 @@ public class HttpSourceActor extends AbstractActor {
             final ActorRef shardRegion,
             final ActorRef emitter,
             final ClusterAggregatorConfiguration configuration) {
-        return Props.create(() -> new HttpSourceActor(shardRegion, emitter, configuration));
+        return Props.create(HttpSourceActor.class, () -> new HttpSourceActor(shardRegion, emitter, configuration));
     }
 
     /**

src/main/java/com/arpnetworking/clusteraggregator/configuration/ClusterAggregatorConfiguration.java

@@ -662,7 +662,7 @@ public boolean validateAggregatorLivelinessTimeout(final Duration aggregatorLive
         private Duration _reaggregationTimeout = Duration.ofMinutes(1);
         @NotNull
         @ValidateWithMethod(methodName = "validateAggregatorLivelinessTimeout", parameterType = Duration.class)
-        private Duration _aggregatorLivelinessTimeout = _reaggregationTimeout.multipliedBy(2);
+        private Duration _aggregatorLivelinessTimeout = Duration.ofMinutes(10);
         @NotNull
         private File _hostPipelineConfiguration;
         @NotNull

src/main/java/com/arpnetworking/clusteraggregator/http/Routes.java

@@ -27,7 +27,7 @@
 import akka.http.javadsl.model.headers.CacheControl;
 import akka.http.javadsl.model.headers.CacheDirectives;
 import akka.japi.function.Function;
-import akka.pattern.PatternsCS;
+import akka.pattern.Patterns;
 import akka.util.ByteString;
 import com.arpnetworking.clusteraggregator.Status;
 import com.arpnetworking.clusteraggregator.models.StatusResponse;
@@ -224,7 +224,7 @@ private CompletionStage<HttpResponse> process(final HttpRequest request) {
     @SuppressWarnings("unchecked")
     private <T> CompletionStage<T> ask(final String actorPath, final Object request, final T defaultValue) {
         return
-                PatternsCS.ask(
+                Patterns.ask(
                         _actorSystem.actorSelection(actorPath),
                         request,
                         Duration.ofSeconds(5))

src/main/java/com/arpnetworking/tsdcore/sinks/HttpSinkActor.java

@@ -18,7 +18,7 @@
 import akka.actor.AbstractActor;
 import akka.actor.Props;
 import akka.http.javadsl.model.StatusCodes;
-import akka.pattern.PatternsCS;
+import akka.pattern.Patterns;
 import com.arpnetworking.logback.annotations.LogValue;
 import com.arpnetworking.metrics.Metrics;
 import com.arpnetworking.metrics.MetricsFactory;
@@ -372,7 +372,7 @@ private void fireRequest(final RequestEntry request, final int attempt) {
                         return new PostFailure(attempt, request, err);
                     }
                 });
-        PatternsCS.pipe(responsePromise, context().dispatcher()).to(self());
+        Patterns.pipe(responsePromise, context().dispatcher()).to(self());
     }
 
     private void scheduleRetry(final RequestEntry requestEntry, final int attempt) {

src/main/java/com/arpnetworking/tsdcore/sinks/circonus/BrokerRefresher.java

@@ -17,7 +17,7 @@
 
 import akka.actor.AbstractActor;
 import akka.actor.Props;
-import akka.pattern.PatternsCS;
+import akka.pattern.Patterns;
 import com.arpnetworking.akka.UniformRandomTimeScheduler;
 import com.arpnetworking.steno.Logger;
 import com.arpnetworking.steno.LoggerFactory;
@@ -111,7 +111,7 @@ private void lookupBrokers() {
         final CompletionStage<Object> promise = _client.getBrokers()
                 .<Object>thenApply(BrokerLookupComplete::new)
                 .exceptionally(BrokerLookupFailure::new);
-        PatternsCS.pipe(promise, context().dispatcher()).to(self());
+        Patterns.pipe(promise, context().dispatcher()).to(self());
     }
 
     private final CirconusClient _client;