external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

test-change log level to LOG

example-fix-logical-rep-cleanup
Ahmet Gedemenli 2022-09-12 13:08:16 +03:00
parent cbb7cff30a
commit 0f230ec98f
3 changed files with 9 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/backend/distributed/operations/repair_shards.c
View File

@ -1182,9 +1182,9 @@ CopyShardTablesViaLogicalReplication(List *shardIntervalList, char *sourceNodeNa
int32 targetNodePort) int32 targetNodePort)
{ {
AcquireLogicalReplicationLock(); AcquireLogicalReplicationLock();
elog(WARNING,"before drop leftovers"); elog(LOG,"before drop leftovers");
DropAllLogicalReplicationLeftovers(SHARD_MOVE); DropAllLogicalReplicationLeftovers(SHARD_MOVE);
elog(WARNING,"after drop leftovers"); elog(LOG,"after drop leftovers");
MemoryContext localContext = AllocSetContextCreate(CurrentMemoryContext, MemoryContext localContext = AllocSetContextCreate(CurrentMemoryContext,
"CopyShardTablesViaLogicalReplication", "CopyShardTablesViaLogicalReplication",

8
src/backend/distributed/replication/multi_logical_replication.c
View File

@ -320,10 +320,10 @@ LogicallyReplicateShards(List *shardList, char *sourceNodeName, int sourceNodePo
* the constraints earlier. * the constraints earlier.
*/ */
CreateForeignConstraintsToReferenceTable(logicalRepTargetList); CreateForeignConstraintsToReferenceTable(logicalRepTargetList);
elog(WARNING,"before drop subs - try"); elog(LOG,"before drop subs - try");
/* we're done, cleanup the publication and subscription */ /* we're done, cleanup the publication and subscription */
DropSubscriptions(logicalRepTargetList); DropSubscriptions(logicalRepTargetList);
elog(WARNING,"after drop subs - try"); elog(LOG,"after drop subs - try");
DropReplicationSlots(sourceConnection, logicalRepTargetList); DropReplicationSlots(sourceConnection, logicalRepTargetList);
DropPublications(sourceConnection, publicationInfoHash); DropPublications(sourceConnection, publicationInfoHash);
@ -349,9 +349,9 @@ LogicallyReplicateShards(List *shardList, char *sourceNodeName, int sourceNodePo
/* reconnect if the connection failed or is waiting for a command */ /* reconnect if the connection failed or is waiting for a command */
// RecreateGroupedLogicalRepTargetsConnections(groupedLogicalRepTargetsHash, // RecreateGroupedLogicalRepTargetsConnections(groupedLogicalRepTargetsHash,
// superUser, databaseName); // superUser, databaseName);
elog(WARNING,"before drop subs - catch"); elog(LOG,"before drop subs - catch");
DropSubscriptions(logicalRepTargetList); DropSubscriptions(logicalRepTargetList);
elog(WARNING,"after drop subs - catch"); elog(LOG,"after drop subs - catch");
/* reconnect if the connection failed or is waiting for a command */ /* reconnect if the connection failed or is waiting for a command */
if (PQstatus(sourceConnection->pgConn) != CONNECTION_OK || if (PQstatus(sourceConnection->pgConn) != CONNECTION_OK ||
PQisBusy(sourceConnection->pgConn)) PQisBusy(sourceConnection->pgConn))

3
src/test/regress/sql/failure_online_move_shard_placement.sql
View File

@ -89,11 +89,14 @@ SELECT master_move_shard_placement(101, 'localhost', :worker_1_port, 'localhost'
SELECT citus.mitmproxy('conn.onQuery(query="^ALTER SUBSCRIPTION .* DISABLE").cancel(' || :pid || ')'); SELECT citus.mitmproxy('conn.onQuery(query="^ALTER SUBSCRIPTION .* DISABLE").cancel(' || :pid || ')');
SELECT master_move_shard_placement(101, 'localhost', :worker_1_port, 'localhost', :worker_2_proxy_port); SELECT master_move_shard_placement(101, 'localhost', :worker_1_port, 'localhost', :worker_2_proxy_port);
SET client_min_messages TO LOG;
-- failure on dropping subscription -- failure on dropping subscription
SELECT citus.mitmproxy('conn.onQuery(query="^ALTER SUBSCRIPTION").after(2).kill()'); SELECT citus.mitmproxy('conn.onQuery(query="^ALTER SUBSCRIPTION").after(2).kill()');
SELECT master_move_shard_placement(101, 'localhost', :worker_1_port, 'localhost', :worker_2_proxy_port); SELECT master_move_shard_placement(101, 'localhost', :worker_1_port, 'localhost', :worker_2_proxy_port);
-- try again
SELECT citus.mitmproxy('conn.onQuery(query="nonexistingquery").kill()'); SELECT citus.mitmproxy('conn.onQuery(query="nonexistingquery").kill()');
SELECT master_move_shard_placement(101, 'localhost', :worker_1_port, 'localhost', :worker_2_proxy_port); SELECT master_move_shard_placement(101, 'localhost', :worker_1_port, 'localhost', :worker_2_proxy_port);
RESET client_min_messages;
-- cancellation on dropping subscription -- cancellation on dropping subscription
SELECT citus.mitmproxy('conn.onQuery(query="^DROP SUBSCRIPTION").cancel(' || :pid || ')'); SELECT citus.mitmproxy('conn.onQuery(query="^DROP SUBSCRIPTION").cancel(' || :pid || ')');