fix(spanner): prevent memory leak and thread blocking in transaction keep-alive

[olavloite]

• Enable setRemoveOnCancelPolicy(true) on KEEP_ALIVE_SERVICE so canceled tasks are immediately purged from DelayedWorkQueue.
• Use a WeakReference<ReadWriteTransaction> in KeepAliveRunnable to prevent scheduled tasks from retaining strong references to transaction instances.
• Use abortedLock.tryLock() in KeepAliveRunnable so the shared executor thread does not block when a transaction is active or retrying.
• Remove duplicate maybeScheduleKeepAlivePing listener registration on keep-alive query completion.
• Add unit tests in ReadWriteTransactionTest verifying task removal on cancel, weak reference retention, non-blocking lock handling, and single ping scheduling on completion.

[gemini-code-assist]

Code Review

This pull request refactors the transaction keep-alive mechanism in ReadWriteTransaction by using a ScheduledThreadPoolExecutor with removeOnCancelPolicy enabled, and converting KeepAliveRunnable into a static nested class with a WeakReference to prevent memory leaks. It also introduces lock-based synchronization using abortedLock to avoid sending keep-alive pings when the transaction is busy. Feedback on this PR highlights a critical issue where the keep-alive loop can silently terminate when the transaction is busy; because keepAliveFuture.isDone() is still false during execution, the synchronous call to reschedule the ping results in a no-op. A fix is suggested to reset keepAliveFuture to null before rescheduling.

[gemini-code-assist]

When tryLock() fails because the transaction is currently busy, calling transaction.maybeScheduleKeepAlivePing() synchronously inside KeepAliveRunnable.run() will result in a no-op. This is because keepAliveFuture.isDone() is still false while the current KeepAliveRunnable task is executing. As a result, the keep-alive ping is never rescheduled, and the keep-alive loop silently terminates.

To fix this, we should synchronize on the transaction instance and set transaction.keepAliveFuture = null before calling maybeScheduleKeepAlivePing(). This allows the rescheduling logic to successfully schedule the next keep-alive ping.

        } else {
          // Transaction is currently busy (executing a statement or retrying).
          // Reschedule keep-alive ping for later since it is active.
          synchronized (transaction) {
            transaction.keepAliveFuture = null;
            transaction.maybeScheduleKeepAlivePing();
          }
        }