Bug#36750146 Ndb : Graceful node shutdown does not avoid event duplicates

Graceful node stop is intended to carefully handover
responsibility for forwarding event streams from the stopping
node to still-running nodes, so that there are no duplicate
events sent to event subscribers as a result of an unexpected
node failure.

However, this was not working correctly which could lead
to issues with event merge and other event consumers even
when performing a graceful node stop.

The problem is fixed and two new tests covering the problem
with Graceful stop are added :

  test_event -n EventConsumer_Graceful
  test_event -n MergeEventConsumer_Graceful

Change-Id: I2effa2e0908c920e48ac235a5e0f3a1fce10413a

Author: frazerclement

storage/ndb/src/kernel/blocks/suma/Suma.cpp

@@ -4758,6 +4758,11 @@ void Suma::doFIRE_TRIG_ORD(Signal *signal, LinearSectionPtr lsptr[3]) {
 
   Uint32 bucket = hashValue % c_no_of_buckets;
   m_max_seen_gci = (gci > m_max_seen_gci ? gci : m_max_seen_gci);
+  /**
+   * Normally a bucket is either in the active or switchover set, or neither.
+   * Exception is during add/drop NG when an active bucket may be in the
+   * switchover set for notifying subscribers of change.
+   */
   if (m_active_buckets.get(bucket) ||
       (m_switchover_buckets.get(bucket) && (check_switchover(bucket, gci)))) {
     jam();
@@ -6207,9 +6212,10 @@ void Suma::execSUMA_HANDOVER_REQ(Signal *signal) {
       nodegroup.clear(nodeId);
       if (get_responsible_node(i) == nodeId &&
           get_responsible_node(i, nodegroup) == getOwnNodeId()) {
-        // I'm will be running this bucket when nodeId shutdown
+        // I will be running this bucket when nodeId shutdown
         jam();
         tmp.set(i);
+        ndbassert(!m_active_buckets.get(i));
         m_switchover_buckets.set(i);
         c_buckets[i].m_switchover_gci = (Uint64(start_gci) << 32) - 1;
         c_buckets[i].m_state |= Bucket::BUCKET_SHUTDOWN_TO;
@@ -6295,6 +6301,7 @@ void Suma::execSUMA_HANDOVER_CONF(Signal *signal) {
               gci, buf, c_no_of_buckets);
     g_eventLogger->info("Suma: handover from node %u gci: %u buckets: %s (%u)",
                         nodeId, gci, buf, c_no_of_buckets);
+    ndbassert(!m_active_buckets.overlaps(tmp));
     m_switchover_buckets.bitOR(tmp);
     ndbrequire(c_startup.m_handover_nodes.get(nodeId));
     c_startup.m_handover_nodes.clear(nodeId);
@@ -6317,6 +6324,7 @@ void Suma::execSUMA_HANDOVER_CONF(Signal *signal) {
               buf, c_no_of_buckets);
     g_eventLogger->info("Suma: handover to node %u gci: %u buckets: %s (%u)",
                         nodeId, gci, buf, c_no_of_buckets);
+    m_active_buckets.bitANDC(tmp);
     m_switchover_buckets.bitOR(tmp);
     c_startup.m_handover_nodes.clear(nodeId);
     DBUG_VOID_RETURN;
@@ -6681,6 +6689,7 @@ void Suma::start_resend(Signal *signal, Uint32 buck) {
   bucket->m_switchover_node = get_responsible_node(buck);
   bucket->m_switchover_gci = max;
 
+  ndbassert(!m_active_buckets.get(buck));
   m_switchover_buckets.set(buck);
 
   signal->theData[0] = SumaContinueB::RESEND_BUCKET;