Switch streaming engine worker harness based on job settings (#35901)

* Switch streaming engine worker harness based on job settings

* Restart StreamingWorkerStatusPages whenever harness type is swithched

* Adding a StreamingWorkerHarnessFactoryOutput class to hold harness and its dependencies

* Using mock in TC instead of java reflection

* Removed null check before setting status page

* Ran spotlessApply

* Updating test to process work

* Adding check directpath experiment is set before switching to FanOutStreamingEngineWorkerHarness

Author: parveensania

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

@@ -310,6 +310,9 @@ public Integer create(PipelineOptions options) {
   class EnableWindmillServiceDirectPathFactory implements DefaultValueFactory<Boolean> {
     @Override
     public Boolean create(PipelineOptions options) {
+      if (ExperimentalOptions.hasExperiment(options, "disable_windmill_service_direct_path")) {
+        return false;
+      }
       return ExperimentalOptions.hasExperiment(options, "enable_windmill_service_direct_path");
     }
   }