Switch streaming engine worker harness based on job settings

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");
     }
   }