Startup sequence improvements follow-up

Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20822

Author: ifesdjeen

src/java/org/apache/cassandra/metrics/DecayingEstimatedHistogramReservoir.java

@@ -586,6 +586,9 @@ public double getMean()
                 sum += bCount * bucketOffsets[i];
             }
 
+            if (elements == 0)
+                return 0d;
+
             return (double) sum / elements;
         }
 

src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java

@@ -232,8 +232,6 @@ private void reportMetadata(ClusterMetadata metadata)
     void reportMetadataInternal(ClusterMetadata metadata)
     {
         Topology topology = AccordTopology.createAccordTopology(metadata);
-        if (topology.isEmpty() && isEmpty())
-            return;
 
         updateMapping(metadata);
         if (Invariants.isParanoid())
@@ -499,7 +497,9 @@ public void reportEpochClosed(Ranges ranges, long epoch)
         if (epoch < minEpoch() || epochs.wasTruncated(epoch))
             return;
 
-        syncPropagator.reportClosed(epoch, mapping.nodes(), ranges);
+        Topology topology = getTopologyForEpoch(epoch);
+        if (topology != null)
+            syncPropagator.reportClosed(epoch, topology.nodes(), ranges);
     }
 
     @VisibleForTesting
@@ -516,7 +516,9 @@ public void reportEpochRetired(Ranges ranges, long epoch)
 
         checkStarted();
         // TODO (expected): ensure we aren't fetching a truncated epoch; otherwise this should be non-null
-        syncPropagator.reportRetired(epoch, mapping.nodes(), ranges);
+        Topology topology = getTopologyForEpoch(epoch);
+        if (topology != null)
+            syncPropagator.reportRetired(epoch, topology.nodes(), ranges);
     }
 
     @Override

src/java/org/apache/cassandra/service/accord/AccordService.java

@@ -298,10 +298,6 @@ public synchronized static AccordService startup(NodeId tcmId)
 
         AccordService as = new AccordService(AccordTopology.tcmIdToAccord(tcmId));
         as.startup();
-        instance = as;
-
-        AccordReplicaMetrics.touch();
-
         replayJournal(as);
 
         as.finishInitialization();
@@ -320,6 +316,8 @@ public synchronized static AccordService startup(NodeId tcmId)
         as.configService.registerListener(as.node.durability());
         as.node.durability().start();
 
+        instance = as;
+        AccordReplicaMetrics.touch();
         WatermarkCollector.fetchAndReportWatermarksAsync(as.configService);
         return as;
     }
@@ -456,6 +454,27 @@ public synchronized void startup()
         // Replay local epochs
         for (ImmutableTopoloyImage image : images)
             configService.reportTopology(image.global);
+
+        // Subscribe to TCM events
+        ChangeListener prevListener = MetadataChangeListener.instance.collector.getAndSet(new ChangeListener()
+        {
+            @Override
+            public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
+            {
+                if (state != State.SHUTDOWN)
+                    configService.maybeReportMetadata(next);
+            }
+        });
+
+        Invariants.require((prevListener instanceof MetadataChangeListener.PreInitStateCollector),
+                           "Listener should have been initialized with Accord pre-init state collector, but was " + prevListener.getClass());
+
+        MetadataChangeListener.PreInitStateCollector preinit = (MetadataChangeListener.PreInitStateCollector) prevListener;
+        for (ClusterMetadata item : preinit.getItems())
+        {
+            if (item.epoch.getEpoch() > Epoch.FIRST.getEpoch())
+                configService.maybeReportMetadata(item);
+        }
     }
 
     /**
@@ -476,27 +495,6 @@ public void finishInitialization()
 
             if (remote != null)
                 remote.forEach(configService::reportTopology, highestKnown + 1, Integer.MAX_VALUE);
-
-            // Subscribe to TCM events
-            ChangeListener prevListener = MetadataChangeListener.instance.collector.getAndSet(new ChangeListener()
-            {
-                @Override
-                public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
-                {
-                    if (state != State.SHUTDOWN)
-                        configService.maybeReportMetadata(next);
-                }
-            });
-
-            Invariants.require((prevListener instanceof MetadataChangeListener.PreInitStateCollector),
-                               "Listener should have been initialized with Accord pre-init state collector, but was " + prevListener.getClass());
-
-            MetadataChangeListener.PreInitStateCollector preinit = (MetadataChangeListener.PreInitStateCollector) prevListener;
-            for (ClusterMetadata item : preinit.getItems())
-            {
-                if (item.epoch.getEpoch() > minEpoch())
-                    configService.maybeReportMetadata(item);
-            }
         }
         catch (InterruptedException e)
         {

test/simulator/main/org/apache/cassandra/simulator/paxos/AccordSimulationRunner.java

@@ -47,7 +47,6 @@ public class AccordSimulationRunner extends SimulationRunner
     @BeforeClass
     public static void beforeAll()
     {
-        CassandraRelevantProperties.JUNIT_STORAGE_COMPATIBILITY_MODE.setString(StorageCompatibilityMode.NONE.toString());
         DatabaseDescriptor.clientInitialization();
     }