### Question:
PressureDatetimeExtractor implements TimestampExtractor { @Override public long extract(ConsumerRecord<Object, Object> record, long previousTimestamp) { return Try .ofFailable(() -> ((PressureAlert) record.value()).getDatetime()) .onFailure((ex) -> logger.error("fail to cast the PressureAlert: ", ex)) .map((stringDatetimeString) -> ZonedDateTime.parse(stringDatetimeString, this.formatter)) .onFailure((ex) -> logger.error("fail to parse the event datetime due to: ", ex)) .map((zonedDatetime) -> zonedDatetime.toInstant().toEpochMilli()) .onFailure((ex) -> logger.error("fail to convert the datetime to instant due to: ", ex)) .orElse(-1L); } PressureDatetimeExtractor(Config config); @Override long extract(ConsumerRecord<Object, Object> record, long previousTimestamp);  }### Answer:
@Test public void extract() { final TestInputTopic<Bytes, PressureAlert> testDriverInputTopic = testDriver.createInputTopic(this.inputTopic, Serdes.Bytes().serializer(), this.pressureSerde.serializer()); List<PressureAlert> inputs = Arrays.asList( new PressureAlert("101", "2019-09-21T05:25:01.+0200", Integer.MAX_VALUE), new PressureAlert("102", "2019-09-21T05:30:02.+0200", Integer.MAX_VALUE), new PressureAlert("103", "2019-09-21T05:45:03.+0200", Integer.MAX_VALUE), new PressureAlert("104", "DEFINITELY-NOT-PARSABLE!!", Integer.MAX_VALUE), new PressureAlert("105", "1500-06-24T09:11:03.+0200", Integer.MAX_VALUE) ); inputs.forEach(pressureAlert -> testDriverInputTopic.pipeInput(null, pressureAlert) ); List<TestRecord<String, PressureAlert>> result = readNOutputs(5); Optional<TestRecord<String, PressureAlert>> resultOne = result.stream().filter(Objects::nonNull).filter(r -> r.value().getId().equals("101")).findFirst(); Optional<TestRecord<String, PressureAlert>> resultTwo = result.stream().filter(Objects::nonNull).filter(r -> r.value().getId().equals("102")).findFirst(); Optional<TestRecord<String, PressureAlert>> resultThree = result.stream().filter(Objects::nonNull).filter(r -> r.value().getId().equals("103")).findFirst(); Optional<TestRecord<String, PressureAlert>> resultFour = result.stream().filter(Objects::nonNull).filter(r -> r.value().getId().equals("104")).findFirst(); Optional<TestRecord<String, PressureAlert>> resultFive = result.stream().filter(Objects::nonNull).filter(r -> r.value().getId().equals("105")).findFirst(); Assert.assertTrue(resultOne.isPresent()); Assert.assertTrue(resultTwo.isPresent()); Assert.assertTrue(resultThree.isPresent()); Assert.assertFalse(resultFour.isPresent()); Assert.assertFalse(resultFive.isPresent()); Assert.assertEquals( formatter.parse("2019-09-21T05:25:01.+0200", Instant::from).toEpochMilli(), resultOne.get().timestamp().longValue() ); Assert.assertEquals( formatter.parse("2019-09-21T05:30:02.+0200", Instant::from).toEpochMilli(), resultTwo.get().timestamp().longValue() ); Assert.assertEquals( formatter.parse("2019-09-21T05:45:03.+0200", Instant::from).toEpochMilli(), resultThree.get().timestamp().longValue() ); }