[Dataflow Streaming] Prevent commit threads from sharing commit streams (#37847)

Since the commit threads were sharing the WindmillStreamPool, the
different commit streams can end up sharing commit streams.

This change gives each commit thread its own WindmillStreamPool and
avoids the commit stream sharing problem.

Author: arunpandianp

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

@@ -464,10 +464,13 @@ private StreamingWorkerHarnessFactoryOutput createSingleSourceWorkerHarness(
     @SuppressWarnings("methodref.receiver.bound")
     WorkCommitter workCommitter =
         StreamingEngineWorkCommitter.builder()
+            // Use a separate stream pool for each committer. This ensures the commit
+            // threads are fully isolated.
             .setCommitWorkStreamFactory(
-                WindmillStreamPool.create(
-                        numCommitThreads, COMMIT_STREAM_TIMEOUT, windmillServer::commitWorkStream)
-                    ::getCloseableStream)
+                () ->
+                    WindmillStreamPool.create(
+                            1, COMMIT_STREAM_TIMEOUT, windmillServer::commitWorkStream)
+                        .getCloseableStream())
             .setCommitByteSemaphore(Commits.maxCommitByteSemaphore())
             .setNumCommitSenders(numCommitThreads)
             .setOnCommitComplete(this::onCompleteCommit)