changes_amit_v60.patch

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 2e2cd0d535..158dbd93c9 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -707,10 +707,10 @@ pa_stop_idle_workers(void)
 }
 
 /*
- * Check if the parallel apply worker is pending due to spooled messages.
+ * Returns true, if there is any pending spooled message, false otherwise.
  */
 static bool
-pa_has_spooled_message_pending()
+pa_is_spooled_message_pending()
 {
 	PartialFileSetState fileset_state;
 
@@ -870,7 +870,7 @@ LogicalParallelApplyLoop(shm_mq_handle *mqh)
 		else if (shmq_res == SHM_MQ_WOULD_BLOCK)
 		{
 			/* Check if changes have been serialized to a file. */
-			if (pa_has_spooled_message_pending())
+			if (pa_is_spooled_message_pending())
 			{
 				pa_spooled_messages();
 			}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index faebceb4c7..59a4366fec 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1558,15 +1558,17 @@ apply_handle_stream_start(StringInfo s)
 
 			pa_send_data(winfo, s->len, s->data, !first_segment);
 
+			/*
+			 * Once we start serializing the changes, the parallel apply worker
+			 * will wait for the leader to release the stream lock until the
+			 * end of the transaction. So, we don't need to release the lock or
+			 * increment the stream count in that case.
+			 */
 			if (!winfo->serialize_changes)
 			{
 				/*
-				 * Unlock the shared object lock when no change was serialized
-				 * so that the parallel apply worker can continue to receive
-				 * changes. No need to take the lock if switched to serialize
-				 * mode because the parallel apply worker should wait for the
-				 * leader to release the stream lock until the end of the
-				 * transaction.
+				 * Unlock the shared object lock so that the parallel apply
+				 * worker can continue to receive changes.
 				 */
 				if (!first_segment)
 					pa_unlock_stream(winfo->shared->xid, AccessExclusiveLock);