private long getRemainingRecords(KafkaConsumer<Integer, String> consumer) {
        final Map<TopicPartition, Long> fullEndOffsets = consumer.endOffsets(new ArrayList<>(consumer.assignment()));
        // if we can't detect any end offset, that means we are still not able to fetch offsets
        if (fullEndOffsets.isEmpty()) {
            return Long.MAX_VALUE;
        }
        return consumer.assignment().stream().mapToLong(partition -> {
            long currentPosition = consumer.position(partition);
            if (fullEndOffsets.containsKey(partition)) {
                return fullEndOffsets.get(partition) - currentPosition;
            } else {
                return 0;
            }
        }).sum();
    }
This class implements a read-process-write application.
examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java