Fix bug in specifying recovery position

knizhnik · knizhnik · commit 4e26210e3c46 · 2016-03-30T14:57:21.000+03:00
diff --git a/contrib/mmts/arbiter.c b/contrib/mmts/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;
diff --git a/contrib/mmts/multimaster.c b/contrib/mmts/multimaster.c
@@ -1749,7 +1749,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;			
@@ -1779,7 +1779,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;
 }
 
diff --git a/contrib/mmts/pglogical_output.c b/contrib/mmts/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;
 	}
 
diff --git a/contrib/mmts/pglogical_receiver.c b/contrib/mmts/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,

Original file line number	Diff line number	Diff line change
`@@ -513,7 +513,7 @@ static void MtmBroadcastMessage(MtmBuffer* txBuffer, MtmTransState* ts)`
`513`	`513`	`int n = 1;`
`514`	`514`	`for (i = 0; i < MtmNodes; i++)`
`515`	`515`	`{`
`516`		`- if (TransactionIdIsValid(ts->xids[i])) {`
	`516`	`+ if (!BIT_CHECK(Mtm->disabledNodeMask, i) && TransactionIdIsValid(ts->xids[i])) {`
`517`	`517`	`Assert(i+1 != MtmNodeId);`
`518`	`518`	`MtmAppendBuffer(txBuffer, ts->xids[i], i, ts);`
`519`	`519`	`n += 1;`
Original file line number	Diff line number	Diff line change
`@@ -489,7 +489,6 @@ pg_decode_origin_filter(LogicalDecodingContext *ctx,`
`489`	`489`	`PGLogicalOutputData *data = ctx->output_plugin_private;`
`490`	`490`
`491`	`491`	`if (!call_txn_filter_hook(data, origin_id)) {`
`492`		`- elog(WARNING, "Record with origin %d is not sent to node %d", origin_id, MtmReplicationNodeId);`
`493`	`492`	`return true;`
`494`	`493`	`}`
`495`	`494`
Original file line number	Diff line number	Diff line change
`@@ -292,7 +292,7 @@ pglogical_receiver_main(Datum main_arg)`
`292`	`292`	`}`
`293`	`293`	`CommitTransactionCommand();`
`294`	`294`
`295`		`- 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')",`
	`295`	`+ 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')",`
`296`	`296`	`args->receiver_slot,`
`297`	`297`	`(uint32) (originStartPos >> 32),`
`298`	`298`	`(uint32) originStartPos,`