Do not change cluster status because of pglogical errors

Author: knizhnik

Diff for: arbiter.c

@@ -401,7 +401,7 @@ static int MtmConnectSocket(char const* host, int port, int max_attempts)
 
 			/* Some node considered that I am dead, so switch to recovery mode */
 			if (BIT_CHECK(resp.disabledNodeMask, MtmNodeId-1)) { 
-				elog(WARNING, "Node %d think that I am dead", resp.node);
+				elog(WARNING, "Node %d thinks that I was dead", resp.node);
 				BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
 				MtmSwitchClusterMode(MTM_RECOVERY);
 			}

Diff for: multimaster.c

@@ -2145,7 +2145,7 @@ MtmOnProcExit(int code, Datum arg)
 	if (MtmReplicationNodeId > 0) { 
 		Mtm->nodes[MtmReplicationNodeId-1].senderPid = -1;
 		MTM_LOG1("WAL-sender to %d is terminated", MtmReplicationNodeId); 
-		MtmOnNodeDisconnect(MtmReplicationNodeId);
+		/* MtmOnNodeDisconnect(MtmReplicationNodeId); */
 	}
 }
 
@@ -2241,7 +2241,7 @@ MtmReplicationShutdownHook(struct PGLogicalShutdownHookArgs* args)
 {
 	if (MtmReplicationNodeId >= 0) { 
 		MTM_LOG1("Logical replication to node %d is stopped", MtmReplicationNodeId); 
-		MtmOnNodeDisconnect(MtmReplicationNodeId);
+		/* MtmOnNodeDisconnect(MtmReplicationNodeId); */
 		MtmReplicationNodeId = -1; /* defuse on_proc_exit hook */
 	}
 }

Diff for: pglogical_receiver.c

@@ -275,7 +275,7 @@ pglogical_receiver_main(Datum main_arg)
 				PQclear(res);
 				ereport(ERROR, (errmsg("%s: Could not create logical slot",
 									   worker_proc)));
-				MtmOnNodeDisconnect(nodeId);
+				/* MtmOnNodeDisconnect(nodeId); */
 				proc_exit(1);
 			}
 		}
@@ -317,7 +317,7 @@ pglogical_receiver_main(Datum main_arg)
 		PQclear(res);
 		ereport(WARNING, (errmsg("%s: Could not start logical replication",
 								 worker_proc)));
-		MtmOnNodeDisconnect(nodeId);
+		/* MtmOnNodeDisconnect(nodeId); */
 		proc_exit(1);
 	}
 	PQclear(res);
@@ -408,7 +408,7 @@ pglogical_receiver_main(Datum main_arg)
 				{
 					ereport(LOG, (errmsg("%s: streaming header too small: %d",
 										 worker_proc, rc)));
-					MtmOnNodeDisconnect(nodeId);
+					/* MtmOnNodeDisconnect(nodeId); */
 					proc_exit(1);
 				}
 				replyRequested = copybuf[pos];
@@ -429,7 +429,7 @@ pglogical_receiver_main(Datum main_arg)
 
 					/* Leave is feedback is not sent properly */
 					if (!sendFeedback(conn, now, nodeId)) {
-						MtmOnNodeDisconnect(nodeId);
+						/* MtmOnNodeDisconnect(nodeId); */
 						proc_exit(1);
 					}
 				}
@@ -439,7 +439,7 @@ pglogical_receiver_main(Datum main_arg)
 			{
 				ereport(LOG, (errmsg("%s: Incorrect streaming header",
 									 worker_proc)));
-				MtmOnNodeDisconnect(nodeId);
+				/* MtmOnNodeDisconnect(nodeId); */
 				proc_exit(1);
 			}
 
@@ -548,7 +548,7 @@ pglogical_receiver_main(Datum main_arg)
 			{
 				ereport(LOG, (errmsg("%s: Incorrect status received... Leaving.",
 									 worker_proc)));
-				MtmOnNodeDisconnect(nodeId);
+				/* MtmOnNodeDisconnect(nodeId); */
 				proc_exit(1);
 			}
 
@@ -557,7 +557,7 @@ pglogical_receiver_main(Datum main_arg)
 			{
 				ereport(LOG, (errmsg("%s: Data remaining on the socket... Leaving.",
 									 worker_proc)));
-				MtmOnNodeDisconnect(nodeId);
+				/* MtmOnNodeDisconnect(nodeId); */
 				proc_exit(1);
 			}
 			continue;
@@ -576,7 +576,7 @@ pglogical_receiver_main(Datum main_arg)
 		{
 			ereport(LOG, (errmsg("%s: Failure while receiving changes...",
 								 worker_proc)));
-			MtmOnNodeDisconnect(nodeId);
+			/* MtmOnNodeDisconnect(nodeId); */
 			proc_exit(1);
 		}
 	}