MonitoringUtil { public static String getGcloudCancelCommand(DataflowPipelineOptions options, String jobId) { String dataflowApiOverridePrefix = ""; String apiUrl = options.getDataflowClient().getBaseUrl(); if (!apiUrl.equals(Dataflow.DEFAULT_BASE_URL)) { dataflowApiOverridePrefix = String.format("%s=%s ", ENDPOINT_OVERRIDE_ENV_VAR, apiUrl); } return String.format( "%s%s jobs --project=%s cancel --region=%s %s", dataflowApiOverridePrefix, GCLOUD_DATAFLOW_PREFIX, options.getProject(), options.getRegion(), jobId); } MonitoringUtil(DataflowClient dataflowClient); List<JobMessage> getJobMessages(String jobId, long startTimestampMs); @Deprecated static String getJobMonitoringPageURL(String projectName, String jobId); static String getJobMonitoringPageURL(String projectName, String regionId, String jobId); static String getGcloudCancelCommand(DataflowPipelineOptions options, String jobId); static State toState(String stateName);  }
@Test public void testDontOverrideEndpointWithDefaultApi() { DataflowPipelineOptions options = PipelineOptionsFactory.create().as(DataflowPipelineOptions.class); options.setProject(PROJECT_ID); options.setRegion(REGION_ID); options.setGcpCredential(new TestCredential()); String cancelCommand = MonitoringUtil.getGcloudCancelCommand(options, JOB_ID); assertEquals( "gcloud dataflow jobs --project=someProject cancel --region=thatRegion 1234", cancelCommand); }
@Test public void testOverridesEndpointWithStagedDataflowEndpoint() { DataflowPipelineOptions options = PipelineOptionsFactory.create().as(DataflowPipelineOptions.class); options.setProject(PROJECT_ID); options.setRegion(REGION_ID); options.setGcpCredential(new TestCredential()); String stagingDataflowEndpoint = "v0neverExisted"; options.setDataflowEndpoint(stagingDataflowEndpoint); String cancelCommand = MonitoringUtil.getGcloudCancelCommand(options, JOB_ID); assertEquals( "CLOUDSDK_API_ENDPOINT_OVERRIDES_DATAFLOW=https: + "gcloud dataflow jobs --project=someProject cancel --region=thatRegion 1234", cancelCommand); }