+ TopicPartition topicPartition =
+ new TopicPartition(record.topic(), record.partition());
+ Message.MessageKey messageKey = Message.MessageKey.of(
+ record.value().getUser(),
+ record.value().getId());
+
+ if (record.offset() >= offsetOfFirstUnseenMessage)