MonitoringUtil { public List<JobMessage> getJobMessages(String jobId, long startTimestampMs) throws IOException { Instant startTimestamp = new Instant(startTimestampMs); ArrayList<JobMessage> allMessages = new ArrayList<>(); String pageToken = null; while (true) { ListJobMessagesResponse response = dataflowClient.listJobMessages(jobId, pageToken); if (response == null || response.getJobMessages() == null) { return allMessages; } for (JobMessage m : response.getJobMessages()) { @Nullable Instant timestamp = fromCloudTime(m.getTime()); if (timestamp == null) { continue; } if (timestamp.isAfter(startTimestamp)) { allMessages.add(m); } } if (response.getNextPageToken() == null) { break; } else { pageToken = response.getNextPageToken(); } } allMessages.sort(new TimeStampComparator()); return allMessages; } 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 testGetJobMessages() throws IOException { DataflowClient dataflowClient = mock(DataflowClient.class); ListJobMessagesResponse firstResponse = new ListJobMessagesResponse(); firstResponse.setJobMessages(new ArrayList<>()); for (long i = 0; i < 100; ++i) { JobMessage message = new JobMessage(); message.setId("message_" + i); message.setTime(TimeUtil.toCloudTime(new Instant(i))); firstResponse.getJobMessages().add(message); } String pageToken = "page_token"; firstResponse.setNextPageToken(pageToken); ListJobMessagesResponse secondResponse = new ListJobMessagesResponse(); secondResponse.setJobMessages(new ArrayList<>()); for (long i = 100; i < 150; ++i) { JobMessage message = new JobMessage(); message.setId("message_" + i); message.setTime(TimeUtil.toCloudTime(new Instant(i))); secondResponse.getJobMessages().add(message); } when(dataflowClient.listJobMessages(JOB_ID, null)).thenReturn(firstResponse); when(dataflowClient.listJobMessages(JOB_ID, pageToken)).thenReturn(secondResponse); MonitoringUtil util = new MonitoringUtil(dataflowClient); List<JobMessage> messages = util.getJobMessages(JOB_ID, -1); assertEquals(150, messages.size()); }