ZookeeperUndeliveredMessageLog implements UndeliveredMessageLog { @Override public Optional<SentMessageTrace> last(TopicName topicName, String subscriptionName) { try { String path = paths.subscriptionPath(topicName, subscriptionName, NODE_NAME); if (exists(path)) { return Optional.of(mapper.readValue(curator.getData().forPath(path), SentMessageTrace.class)); } else { return Optional.empty(); } } catch (Exception e) { throw new InternalProcessingException( format("Could not read latest undelivered message for topic: %s and subscription: %s .", topicName.qualifiedName(), subscriptionName), e); } } ZookeeperUndeliveredMessageLog(CuratorFramework curator, ZookeeperPaths zookeeperPaths, ObjectMapper mapper); @Override void add(SentMessageTrace message); @Override Optional<SentMessageTrace> last(TopicName topicName, String subscriptionName); @Override void persist();  }
@Test public void shouldReturnAbsentIfThereAreNoUndeliveredMessagesForGivenSubscription() { Optional<SentMessageTrace> result = log.last(new TopicName("unknown", "topic"), "subscription"); assertThat(result.isPresent()).isFalse(); }