Fix recovery completed message handling

Author: knizhnik

multimaster.c

@@ -3293,6 +3293,7 @@ void MtmRecoverNode(int nodeId)
 	{
 		Assert(BIT_CHECK(Mtm->disabledNodeMask, nodeId-1));
 		BIT_CLEAR(Mtm->stoppedNodeMask, nodeId-1);
+		BIT_CLEAR(Mtm->stalledNodeMask, nodeId-1);
 	}
 	MtmUnlock();
 
@@ -3551,8 +3552,8 @@ MtmReplicationRowFilterHook(struct PGLogicalRowFilterArgs* args)
  * Filter received transactions at destination side.
  * This function is executed by receiver, 
  * so there are no race conditions and it is possible to update nodes[i].restartLSN without lock. 
- * It is more efficient to filter records at senders size (done by MtmReplicationTxnFilterHook) to avoid sending useless data through network. But asynchronous nature of 
- * logical replications makes it not possible to guarantee (at least I failed to do it)
+ * It is more efficient to filter records at senders size (done by MtmReplicationTxnFilterHook) to avoid sending useless data through network. 
+ * But asynchronous nature of logical replications makes it not possible to guarantee (at least I failed to do it)
  * that replica do not receive deteriorated data.
  */
 bool MtmFilterTransaction(char* record, int size)

pglogical_apply.c

@@ -1119,7 +1119,7 @@ void MtmExecutor(void* work, size_t size)
 			case 'Z':
 			{
 				MtmRecoveryCompleted();
-				continue;
+				break;
 			}
             default:
                 elog(ERROR, "unknown action of type %c", action);

pglogical_proto.c

@@ -182,7 +182,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 {
     uint8 event = 0;
 
-	MTM_LOG1("%d: pglogical_write_commit XID=%ld node=%d restart_decoding_lsn=%lx first_lsn=%lx end_lsn=%lx confirmed_flush=%lx", 
+	MTM_LOG2("%d: pglogical_write_commit XID=%ld node=%d restart_decoding_lsn=%lx first_lsn=%lx end_lsn=%lx confirmed_flush=%lx", 
 			 MyProcPid, (long)txn->xid, MtmReplicationNodeId, txn->restart_decoding_lsn, txn->first_lsn, txn->end_lsn, MyReplicationSlot->data.confirmed_flush);
 
 

tests2/test_recovery.py

@@ -87,7 +87,7 @@ def tearDownClass(self):
         print('tearDown')
         self.client.stop()
         # XXX: check nodes data identity here
-        subprocess.check_call(['docker-compose','down'])
+#        subprocess.check_call(['docker-compose','down'])
 
     def setUp(self):
         warnings.simplefilter("ignore", ResourceWarning)