[#28889] xClusterDDLRepl: Fix IncrementalSafeTimeBumpWithDdlQueueStepdowns in ASAN

Summary:
After introducing advisory locks to ddl_queue_handler, only one handler can be active at once. This test creates a handler, pauses (which deletes the handler), then resumes (which recreates the handler). When the handler is deleted, it takes a bit for the pg session to shutdown, which means that the advisory lock is held on for an extended period of time. During this period the next handler can't grab the advisory lock, so it can't make progress. In build types that are more time sensitive (eg ASAN) , this causes checks to occur before the new handler is ready.

Fix is to try to release all advisory locks when shutting down the ddl queue handler.
Jira: DB-18611

Test Plan:
```
ybd asan --cxx-test integration-tests_xcluster_ddl_replication-test --gtest_filter XClusterDDLReplicationTest.IncrementalSafeTimeBumpWithDdlQueueStepdowns -n 40
```

Reviewers: hsunder, xCluster

Reviewed By: hsunder

Subscribers: ybase

Differential Revision: https://phorge.dev.yugabyte.com/D47362

Author: hulien22

src/yb/tserver/xcluster_ddl_queue_handler.cc

@@ -310,6 +310,17 @@ XClusterDDLQueueHandler::XClusterDDLQueueHandler(
 
 XClusterDDLQueueHandler::~XClusterDDLQueueHandler() {}
 
+void XClusterDDLQueueHandler::Shutdown() {
+  if (pg_conn_ && FLAGS_ysql_yb_enable_advisory_locks &&
+      FLAGS_xcluster_ddl_queue_advisory_lock_key != 0) {
+    // Optimistically unlock the advisory lock so we don't have to wait for the connection to close.
+    auto s = pg_conn_->Execute(Format("SELECT pg_advisory_unlock_all()"));
+    // Alright if we fail here, log an error and wait for the connection to close normally.
+    WARN_NOT_OK(s, "Encountered error unlocking advisory lock for xCluster DDL queue handler");
+    pg_conn_.reset();
+  }
+}
+
 Status XClusterDDLQueueHandler::ExecuteCommittedDDLs() {
   SCHECK(safe_time_batch_, InternalError, "Safe time batch is not initialized");
   if (!safe_time_batch_->IsComplete()) {

src/yb/tserver/xcluster_ddl_queue_handler.h

@@ -79,6 +79,8 @@ class XClusterDDLQueueHandler {
       ConnectToPostgresFunc connect_to_pg_func, UpdateSafeTimeFunc update_safe_time_func);
   virtual ~XClusterDDLQueueHandler();
 
+  void Shutdown();
+
   // This function is called before the poller calls GetChanges. This will detect if we are in the
   // middle of a executing a DDL batch and complete it.
   Status ProcessPendingBatchIfExists();

src/yb/tserver/xcluster_poller.cc

@@ -215,6 +215,10 @@ void XClusterPoller::CompleteShutdown() {
   if (output_client_) {
     output_client_->CompleteShutdown();
   }
+  if (ddl_queue_handler_) {
+    // Only clean up the ddl queue handler after all other tasks have completed.
+    ddl_queue_handler_->Shutdown();
+  }
 
   XClusterAsyncExecutor::CompleteShutdown();