[Dataflow Streaming] Create a separate option to control commit threads with direct path. (#37848)

The number of commit threads will be per backend windmill worker.

Author: arunpandianp

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowStreamingPipelineOptions.java

@@ -100,6 +100,13 @@ public interface DataflowStreamingPipelineOptions extends PipelineOptions {
 
   void setWindmillServiceCommitThreads(Integer value);
 
+  @Description(
+      "Number of commit threads per backend windmill worker in streaming engine direct path mode.")
+  @Default.Integer(1)
+  Integer getWindmillServiceDirectPathCommitThreads();
+
+  void setWindmillServiceDirectPathCommitThreads(Integer value);
+
   @Description(
       "Frequency at which active work should be reported back to Windmill, in millis. "
           + "The first refresh will occur after at least this much time has passed since "

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

@@ -419,7 +419,8 @@ private StreamingWorkerHarnessFactoryOutput createFanOutStreamingEngineWorkerHar
                     .setCommitByteSemaphore(maxCommitByteSemaphore)
                     .setBackendWorkerToken(commitWorkStream.backendWorkerToken())
                     .setOnCommitComplete(this::onCompleteCommit)
-                    .setNumCommitSenders(Math.max(options.getWindmillServiceCommitThreads(), 1))
+                    .setNumCommitSenders(
+                        Math.max(options.getWindmillServiceDirectPathCommitThreads(), 1))
                     .setCommitWorkStreamFactory(
                         () -> CloseableStream.create(commitWorkStream, () -> {}))
                     .build(),