Random subscription 021_twophase test failure on kestrel

From 4f75511e9866ef29f79789e5be0a31f0ee5786e7 Mon Sep 17 00:00:00 2001
From: Vignesh <vignesh21@gmail.com>
Date: Fri, 23 May 2025 20:40:40 +0530
Subject: [PATCH] Fix random 021_twophase test failure

This issue occurred because the test dropped the subscription
tap_sub before confirming it had replayed the commit for the
previously prepared transaction 'mygid'. The test only waited
for the second subscription, tap_sub_copy, to catch up. As a
result, tap_sub had not yet applied the commit, leaving the
prepared transaction uncommitted on the subscriber.
---
 src/test/subscription/t/021_twophase.pl | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl
index 61c427aed21..b8e4242d1f1 100644
--- a/src/test/subscription/t/021_twophase.pl
+++ b/src/test/subscription/t/021_twophase.pl
@@ -373,7 +373,14 @@ $result =
   $node_publisher->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
 is($result, qq(6), 'publisher inserted data');
 
+# Wait for both subscribers to catchup
 $node_publisher->wait_for_catchup($appname_copy);
+$node_publisher->wait_for_catchup($appname);
+
+# Make sure there are no prepared transactions on the subscriber
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT count(*) FROM pg_prepared_xacts;");
+is($result, qq(0), 'should be no prepared transactions on subscriber');
 
 $result =
   $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
-- 
2.43.0

diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index cc03f0706e9..981a8e45003 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -38,6 +38,7 @@
 #include "replication/message.h"
 #include "replication/reorderbuffer.h"
 #include "replication/snapbuild.h"
+#include "replication/slot.h"
 #include "storage/standbydefs.h"
 
 /* individual record(group)'s handlers */
@@ -682,6 +683,10 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	RepOriginId origin_id = XLogRecGetOrigin(buf->record);
 	int			i;
 
+	if (parsed->twophase_xid && strcmp(parsed->twophase_gid, "mygid") == 0 &&
+		strcmp(NameStr(MyReplicationSlot->data.name), "tap_sub") == 0)
+		sleep(3);
+
 	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
 	{
 		origin_lsn = parsed->origin_lsn;