TaskInfoFactory { public static Task parse(Protos.TaskInfo taskInfo, Protos.TaskStatus taskStatus, Clock clock) { Properties data = new Properties(); try { data.load(taskInfo.getData().newInput()); } catch (IOException e) { throw new RuntimeException("Failed to parse properties", e); } String hostName = Optional.ofNullable(data.getProperty("hostname")).orElseGet(() -> { LOGGER.error("Hostname is empty. Reported IP addresses will be incorrect."); return ""; }); String ipAddress = data.getProperty("ipAddress", hostName); final ZonedDateTime startedAt = Optional.ofNullable(data.getProperty("startedAt")) .map(s -> s.endsWith("...") ? s.substring(0, 29) : s) .map(ZonedDateTime::parse) .orElseGet(clock::nowUTC) .withZoneSameInstant(ZoneOffset.UTC); if (!taskInfo.getDiscovery().isInitialized()) { throw new IndexOutOfBoundsException("TaskInfo has no discovery information."); } return new Task( hostName, taskInfo.getTaskId().getValue(), taskStatus == null ? Protos.TaskState.TASK_STAGING : taskStatus.getState(), startedAt, new InetSocketAddress(ipAddress, taskInfo.getDiscovery().getPorts().getPorts(Discovery.CLIENT_PORT_INDEX).getNumber()), new InetSocketAddress(ipAddress, taskInfo.getDiscovery().getPorts().getPorts(Discovery.TRANSPORT_PORT_INDEX).getNumber()) ); } TaskInfoFactory(ClusterState clusterState); Protos.TaskInfo createTask(Configuration configuration, FrameworkState frameworkState, Protos.Offer offer, Clock clock); ByteString toData(String hostname, String ipAddress, ZonedDateTime zonedDateTime); static Task parse(Protos.TaskInfo taskInfo, Protos.TaskStatus taskStatus, Clock clock); static final String TASK_DATE_FORMAT; }
@Test public void canParseTask() throws Exception { final ZonedDateTime nowUTC = ZonedDateTime.now(ZoneOffset.UTC); when(clock.nowUTC()).thenReturn(nowUTC.minusYears(100)); final Protos.TaskID taskId = Protos.TaskID.newBuilder().setValue("TaskID").build(); final Protos.TaskInfo taskInfo = createTaskInfo(taskId, createData(Optional.of(nowUTC))); final Protos.TaskStatus taskStatus = Protos.TaskStatus.newBuilder() .setTaskId(taskId) .setState(Protos.TaskState.TASK_STAGING) .build(); final Task task = TaskInfoFactory.parse(taskInfo, taskStatus, clock); assertEquals(nowUTC, task.getStartedAt()); verify(clock, never()).nowUTC(); }
@Test public void canParseTaskWithoutTimestamp() throws Exception { final ZonedDateTime nowUTC = ZonedDateTime.now(ZoneOffset.UTC); when(clock.nowUTC()).thenReturn(nowUTC); final Protos.TaskID taskId = Protos.TaskID.newBuilder().setValue("TaskID").build(); final Protos.TaskInfo taskInfo = createTaskInfo(taskId, createData(Optional.<ZonedDateTime>empty())); final Protos.TaskStatus taskStatus = Protos.TaskStatus.newBuilder() .setTaskId(taskId) .setState(Protos.TaskState.TASK_STAGING) .build(); final Task task = TaskInfoFactory.parse(taskInfo, taskStatus, clock); assertEquals(nowUTC, task.getStartedAt()); }
@Test public void canParseTaskCappedTimestamp() throws Exception { final ZonedDateTime nowUTC = ZonedDateTime.now(ZoneOffset.UTC); when(clock.nowUTC()).thenReturn(nowUTC); final Protos.TaskID taskId = Protos.TaskID.newBuilder().setValue("TaskID").build(); Properties data = new Properties(); data.put("hostname", "hostname"); data.put("ipAddress", "ip address"); data.put("startedAt", nowUTC.withZoneSameInstant(ZoneId.of("Europe/Paris")).toString()); StringWriter writer = new StringWriter(); data.list(new PrintWriter(writer)); final Protos.TaskInfo taskInfo = createTaskInfo(taskId, ByteString.copyFromUtf8(writer.getBuffer().toString())); final Protos.TaskStatus taskStatus = Protos.TaskStatus.newBuilder() .setTaskId(taskId) .setState(Protos.TaskState.TASK_STAGING) .build(); final Task task = TaskInfoFactory.parse(taskInfo, taskStatus, clock); assertEquals(nowUTC, task.getStartedAt()); }