Fix bug in specifying recovery position

Author: knizhnik

arbiter.c

@@ -513,7 +513,7 @@ static void MtmBroadcastMessage(MtmBuffer* txBuffer, MtmTransState* ts)
 	int n = 1;
 	for (i = 0; i < MtmNodes; i++)
 	{
-		if (TransactionIdIsValid(ts->xids[i])) { 
+		if (!BIT_CHECK(Mtm->disabledNodeMask, i) && TransactionIdIsValid(ts->xids[i])) { 
 			Assert(i+1 != MtmNodeId);
 			MtmAppendBuffer(txBuffer, ts->xids[i], i, ts);
 			n += 1;

multimaster.c

@@ -1748,7 +1748,7 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 	}
 	MtmLock(LW_EXCLUSIVE);
 	if (isRecoverySession) {
-		elog(WARNING, "Node %d start recovery of node %d", MtmNodeId, MtmReplicationNodeId);
+		elog(WARNING, "%d: Node %d start recovery of node %d", MyProcPid, MtmNodeId, MtmReplicationNodeId);
 		if (!BIT_CHECK(Mtm->disabledNodeMask,  MtmReplicationNodeId-1)) {
 			BIT_SET(Mtm->disabledNodeMask,  MtmReplicationNodeId-1);
 			Mtm->nNodes -= 1;			
@@ -1778,7 +1778,6 @@ MtmReplicationTxnFilterHook(struct PGLogicalTxnFilterArgs* args)
 	bool res = Mtm->status != MTM_RECOVERY
 		&& (args->origin_id == InvalidRepOriginId 
 			|| MtmIsRecoveredNode(MtmReplicationNodeId));
-	MTM_TRACE("%d: MtmReplicationTxnFilterHook->%d\n", MyProcPid, res);
 	return res;
 }
 

pglogical_output.c

@@ -489,7 +489,6 @@ pg_decode_origin_filter(LogicalDecodingContext *ctx,
 	PGLogicalOutputData *data = ctx->output_plugin_private;
 
 	if (!call_txn_filter_hook(data, origin_id)) { 
-		elog(WARNING, "Record with origin %d is not sent to node %d", origin_id, MtmReplicationNodeId);
 		return true;
 	}
 

pglogical_receiver.c

@@ -292,7 +292,7 @@ pglogical_receiver_main(Datum main_arg)
 	}
 	CommitTransactionCommand();
 
-	appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %u/%u (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d', \"forward_changesets\" '1', \"mtm_replication_mode\" '%s')",
+	appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %x/%x (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d', \"forward_changesets\" '1', \"mtm_replication_mode\" '%s')",
 					  args->receiver_slot,
 					  (uint32) (originStartPos >> 32),
 					  (uint32) originStartPos,