just disable

Author: kelvich

multimaster.c

@@ -3563,13 +3563,13 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 				ulong64 recoveredLSN;
 				sscanf(strVal(elem->arg), "%llx", &recoveredLSN);
 				MTM_LOG1("Recovered position of node %d is %llx", MtmReplicationNodeId, recoveredLSN);
-				if (Mtm->nodes[MtmReplicationNodeId-1].restartLSN < recoveredLSN) {
-					MTM_LOG1("Advance restartLSN for node %d from %llx to %llx (MtmReplicationStartupHook)",
-							 MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, recoveredLSN);
-					// Assert(Mtm->nodes[MtmReplicationNodeId-1].restartLSN == INVALID_LSN
-					// 	   || recoveredLSN < Mtm->nodes[MtmReplicationNodeId-1].restartLSN + MtmMaxRecoveryLag);
-					Mtm->nodes[MtmReplicationNodeId-1].restartLSN = recoveredLSN;
-				}
+				// if (Mtm->nodes[MtmReplicationNodeId-1].restartLSN < recoveredLSN) {
+				// 	MTM_LOG1("Advance restartLSN for node %d from %llx to %llx (MtmReplicationStartupHook)",
+				// 			 MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, recoveredLSN);
+				// 	// Assert(Mtm->nodes[MtmReplicationNodeId-1].restartLSN == INVALID_LSN
+				// 	// 	   || recoveredLSN < Mtm->nodes[MtmReplicationNodeId-1].restartLSN + MtmMaxRecoveryLag);
+				// 	Mtm->nodes[MtmReplicationNodeId-1].restartLSN = recoveredLSN;
+				// }
 			} else {
 				MTM_ELOG(ERROR, "Recovered position is not specified");
 			}
@@ -3787,10 +3787,10 @@ bool MtmFilterTransaction(char* record, int size)
 		break;
 	}
 
-	if (event != PGLOGICAL_COMMIT_PREPARED)
-		return false;
+	return false;
 
 	restart_lsn = origin_node == MtmReplicationNodeId ? end_lsn : origin_lsn;
+
 	if (Mtm->nodes[origin_node-1].restartLSN < restart_lsn) {
 		MTM_LOG2("[restartlsn] node %d: %llx -> %llx (MtmFilterTransaction)", MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, restart_lsn);
 		if (event != PGLOGICAL_PREPARE) {

pglogical_apply.c

@@ -466,17 +466,18 @@ process_remote_message(StringInfo s)
 				Assert(msg->origin_lsn == INVALID_LSN);
 				msg->origin_lsn = MtmSenderWalEnd;
 			}
-			if (Mtm->nodes[origin_node-1].restartLSN < msg->origin_lsn) { 
-				MTM_LOG1("Receive logical abort message for transaction %s from node %d: %llx < %llx", msg->gid, origin_node, Mtm->nodes[origin_node-1].restartLSN, msg->origin_lsn);
-				Mtm->nodes[origin_node-1].restartLSN = msg->origin_lsn;
-				replorigin_session_origin_lsn = msg->origin_lsn; 				
-				MtmRollbackPreparedTransaction(origin_node, msg->gid);
-			} else { 
-				if (msg->origin_lsn != INVALID_LSN) { 
-					MTM_LOG1("Ignore rollback of transaction %s from node %d because it's LSN %llx <= %llx", 
-							 msg->gid, origin_node, msg->origin_lsn, Mtm->nodes[origin_node-1].restartLSN);
-				}
-			}
+			MtmRollbackPreparedTransaction(origin_node, msg->gid);
+			// if (Mtm->nodes[origin_node-1].restartLSN < msg->origin_lsn) { 
+			// 	MTM_LOG1("Receive logical abort message for transaction %s from node %d: %llx < %llx", msg->gid, origin_node, Mtm->nodes[origin_node-1].restartLSN, msg->origin_lsn);
+			// 	Mtm->nodes[origin_node-1].restartLSN = msg->origin_lsn;
+			// 	replorigin_session_origin_lsn = msg->origin_lsn; 				
+			// 	MtmRollbackPreparedTransaction(origin_node, msg->gid);
+			// } else { 
+			// 	if (msg->origin_lsn != INVALID_LSN) { 
+			// 		MTM_LOG1("Ignore rollback of transaction %s from node %d because it's LSN %llx <= %llx", 
+			// 				 msg->gid, origin_node, msg->origin_lsn, Mtm->nodes[origin_node-1].restartLSN);
+			// 	}
+			// }
 			standalone = true;
 			break;
 		}

pglogical_receiver.c

@@ -349,10 +349,10 @@ pglogical_receiver_main(Datum main_arg)
 			resetPQExpBuffer(query);
 			Mtm->nodes[nodeId-1].manualRecovery = false;
 		} else {
-			if (Mtm->nodes[nodeId-1].restartLSN < originStartPos) {
-				MTM_LOG1("Advance restartLSN for node %d: from %llx to %llx (pglogical_receiver_main)", nodeId, Mtm->nodes[nodeId-1].restartLSN, originStartPos);
-				Mtm->nodes[nodeId-1].restartLSN = originStartPos;
-			}
+			// if (Mtm->nodes[nodeId-1].restartLSN < originStartPos) {
+			// 	MTM_LOG1("Advance restartLSN for node %d: from %llx to %llx (pglogical_receiver_main)", nodeId, Mtm->nodes[nodeId-1].restartLSN, originStartPos);
+			// 	Mtm->nodes[nodeId-1].restartLSN = originStartPos;
+			// }
 			MTM_LOG1("Restart logical receiver at position %llx from node %d", originStartPos, nodeId);
 		}