Bug #20742 Assertion in drop of ndb binlog events after node restart

- make sure TE_ACTIVE event comes when node restarts so that ref counting is correct for drop of event

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

@@ -2667,7 +2667,8 @@ Suma::reportAllSubscribers(Signal *signal,
 {
   SubTableData * data  = (SubTableData*)signal->getDataPtrSend();
 
-  if (table_event == NdbDictionary::Event::_TE_SUBSCRIBE)
+  if (table_event == NdbDictionary::Event::_TE_SUBSCRIBE &&
+      !c_startup.m_restart_server_node_id)
   {
     data->gci            = m_last_complete_gci + 1;
     data->tableId        = subPtr.p->m_tableId;

storage/ndb/src/ndbapi/ClusterMgr.cpp

@@ -396,6 +396,8 @@ ClusterMgr::execNF_COMPLETEREP(const Uint32 * theData){
 
 void
 ClusterMgr::reportConnected(NodeId nodeId){
+  DBUG_ENTER("ClusterMgr::reportConnected");
+  DBUG_PRINT("info", ("nodeId: %u", nodeId));
   /**
    * Ensure that we are sending heartbeat every 100 ms
    * until we have got the first reply from NDB providing
@@ -421,6 +423,7 @@ ClusterMgr::reportConnected(NodeId nodeId){
   theNode.nfCompleteRep = true;
   
   theFacade.ReportNodeAlive(nodeId);
+  DBUG_VOID_RETURN;
 }
 
 void

storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp

@@ -1521,6 +1521,46 @@ NdbEventBuffer::complete_outof_order_gcis()
   ndbout_c("complete_outof_order_gcis: m_latestGCI: %lld", m_latestGCI);
 }
 
+void
+NdbEventBuffer::report_node_connected(Uint32 node_id)
+{
+  NdbEventOperation* op= m_ndb->getEventOperation(0);
+  if (op == 0)
+    return;
+
+  DBUG_ENTER("NdbEventBuffer::report_node_connected");
+  SubTableData data;
+  LinearSectionPtr ptr[3];
+  bzero(&data, sizeof(data));
+  bzero(ptr, sizeof(ptr));
+
+  data.tableId = ~0;
+  data.operation = NdbDictionary::Event::_TE_ACTIVE;
+  data.req_nodeid = (Uint8)node_id;
+  data.ndbd_nodeid = (Uint8)node_id;
+  data.logType = SubTableData::LOG;
+  data.gci = m_latestGCI + 1;
+  /**
+   * Insert this event for each operation
+   */
+  {
+    // no need to lock()/unlock(), receive thread calls this
+    NdbEventOperationImpl* impl = &op->m_impl;
+    do if (!impl->m_node_bit_mask.isclear())
+    {
+      data.senderData = impl->m_oid;
+      insertDataL(impl, &data, ptr);
+    } while((impl = impl->m_next));
+    for (impl = m_dropped_ev_op; impl; impl = impl->m_next)
+      if (!impl->m_node_bit_mask.isclear())
+      {
+        data.senderData = impl->m_oid;
+        insertDataL(impl, &data, ptr);
+      }
+  }
+  DBUG_VOID_RETURN;
+}
+
 void
 NdbEventBuffer::report_node_failure(Uint32 node_id)
 {

storage/ndb/src/ndbapi/NdbEventOperationImpl.hpp

@@ -422,6 +422,7 @@ public:
   void execSUB_GCP_COMPLETE_REP(const SubGcpCompleteRep * const rep);
   void complete_outof_order_gcis();
   
+  void report_node_connected(Uint32 node_id);
   void report_node_failure(Uint32 node_id);
   void completeClusterFailed();
 

storage/ndb/src/ndbapi/Ndbif.cpp

@@ -177,6 +177,7 @@ Ndb::executeMessage(void* NdbObject,
 
 void Ndb::connected(Uint32 ref)
 {
+// cluster connect, a_node == own reference
   theMyRef= ref;
   Uint32 tmpTheNode= refToNode(ref);
   Uint64 tBlockNo= refToBlock(ref);
@@ -209,16 +210,30 @@ void Ndb::connected(Uint32 ref)
   theNode= tmpTheNode; // flag that Ndb object is initialized
 }
 
+void Ndb::report_node_connected(Uint32 nodeId)
+{
+  if (theEventBuffer)
+  {
+    // node connected
+    // eventOperations in the ndb object should be notified
+    theEventBuffer->report_node_connected(nodeId);
+  }
+}
+
 void
 Ndb::statusMessage(void* NdbObject, Uint32 a_node, bool alive, bool nfComplete)
 {
   DBUG_ENTER("Ndb::statusMessage");
+  DBUG_PRINT("info", ("a_node: %u  alive: %u  nfComplete: %u",
+                      a_node, alive, nfComplete));
   Ndb* tNdb = (Ndb*)NdbObject;
   if (alive) {
     if (nfComplete) {
+      // cluster connect, a_node == own reference
       tNdb->connected(a_node);
       DBUG_VOID_RETURN;
     }//if
+    tNdb->report_node_connected(a_node);
   } else {
     if (nfComplete) {
       tNdb->report_node_failure_completed(a_node);

storage/ndb/src/ndbapi/TransporterFacade.cpp

@@ -794,6 +794,8 @@ TransporterFacade::connected()
 void
 TransporterFacade::ReportNodeDead(NodeId tNodeId)
 {
+  DBUG_ENTER("TransporterFacade::ReportNodeDead");
+  DBUG_PRINT("enter",("nodeid= %d", tNodeId));
   /**
    * When a node fails we must report this to each Ndb object. 
    * The function that is used for communicating node failures is called.
@@ -810,6 +812,7 @@ TransporterFacade::ReportNodeDead(NodeId tNodeId)
       (*RegPC) (obj, tNodeId, false, false);
     }
   }
+  DBUG_VOID_RETURN;
 }
 
 void