extra check to skip replication after logical message was written to WAL

Author: kelvich

contrib/mmts/multimaster.c

@@ -1660,7 +1660,7 @@ bool MtmRefreshClusterStatus(bool nowait, int testNodeId)
 			/* Interrupt voting for active transaction and abort them */
 			for (ts = Mtm->transListHead; ts != NULL; ts = ts->next) { 
 				MTM_LOG3("Active transaction gid='%s', coordinator=%d, xid=%d, status=%d, gtid.xid=%d",
-						 ts->gid, ts->gtid.nхode, ts->xid, ts->status, ts->gtid.xid);
+						 ts->gid, ts->gtid.node, ts->xid, ts->status, ts->gtid.xid);
 				if (MtmIsCoordinator(ts)) { 
 					if (!ts->votingCompleted && disabled != 0 && ts->status != TRANSACTION_STATUS_ABORTED) {
 						MtmAbortTransaction(ts);
@@ -3826,6 +3826,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 	{
 		MTM_LOG1("Xact accessed temp table, stopping replication");
 		MtmTx.isDistributed = false; /* Skip */
+		MtmTx.snapshot = INVALID_CSN;
 	}
 
 }

contrib/mmts/pglogical_proto.c

@@ -112,12 +112,14 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 
 	if (!isRecovery && csn == INVALID_CSN) { 
 		MtmIsFilteredTxn = true;
-	} else { 
+		MTM_LOG3("%d: pglogical_write_begin XID=%d filtered", MyProcPid, txn->xid);
+	} else {
+		MTM_LOG3("%d: pglogical_write_begin XID=%d sent", MyProcPid, txn->xid);
+		MtmIsFilteredTxn = false;
 		pq_sendbyte(out, 'B');		/* BEGIN */
 		pq_sendint(out, MtmNodeId, 4);
 		pq_sendint(out, isRecovery ? InvalidTransactionId : txn->xid, 4);
 		pq_sendint64(out, csn);
-		MtmIsFilteredTxn = false;
 		MtmTransactionRecords = 0;
 	}
 }
@@ -126,6 +128,12 @@ static void
 pglogical_write_message(StringInfo out,
 					const char *prefix, Size sz, const char *message)
 {
+	if (MtmIsFilteredTxn)
+	{
+		MTM_LOG3("%d: pglogical_write_message filtered", MyProcPid);
+		return;
+	}
+
 	pq_sendbyte(out, 'G');
 	pq_sendbytes(out, message, sz);
 	pq_sendbyte(out, '\0');