+
+ props = new Properties();
+ props.put(BOOTSTRAP_SERVERS_CONFIG, properties.bootstrapServers);
+ props.put(GROUP_ID_CONFIG, "outbox");
+ props.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ props.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+
+ KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
+ consumer.subscribe(Arrays.asList(this.topic));
+ List<PartitionInfo> partitions = consumer.listTopics().get(this.topic);
+ Set<TopicPartition> assignment = new HashSet<>();
+ for (PartitionInfo info : partitions)
+ {
+ LOG.debug("Found {}/{} (ISR: {})", info.topic(), info.partition(), info.inSyncReplicas());
+ assignment.add(new TopicPartition(info.topic(), info.partition()));
+ }
+
+ LOG.info("Using topic {} with {} partitions", topic, partitions);
+
+ long[] currentOffsets = new long[partitions.size()];
+ for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : consumer.committed(assignment).entrySet())
+ {
+ LOG.info("Found current offset {} for partition {}", entry.getValue(), entry.getKey());
+ currentOffsets[entry.getKey().partition()] = entry.getValue().offset() - 1l;
+ }
+ LOG.info("Current offsets: {}", currentOffsets);
+
+ long[] endOffsets = new long[partitions.size()];
+ for (Map.Entry<TopicPartition, Long> entry : consumer.endOffsets(assignment).entrySet())
+ {
+ LOG.info("Found next offset {} for partition {}", entry.getValue(), entry.getKey());
+ endOffsets[entry.getKey().partition()] = entry.getValue() - 1l;
+ }
+ LOG.info("End-offsets: {}", endOffsets);
+
+ int deleted = 0;
+ while(!Arrays.equals(currentOffsets, endOffsets))
+ {
+ ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
+ LOG.debug("Fetched {} records", records.count());
+ records.forEach(record ->
+ {
+ long recordSequenceNumber = Longs.fromByteArray(record.headers().lastHeader(HEADER).value());
+ LOG.debug(
+ "Found message #{} on offset {} of partition {}",
+ recordSequenceNumber,
+ record.offset(),
+ record.partition());
+ send.add(recordSequenceNumber);
+ currentOffsets[record.partition()] = record.offset();
+ });
+ deleted += cleanUp();
+ LOG.debug("Current offsets: {}", currentOffsets);
+ }
+
+ LOG.info("Cleaned up {} already send entries from outbox table", deleted);
+
+ consumer.close();
+
+ this.clock = clock;
+ this.cleanupInterval = properties.cleanupInterval;
+ this.nextCleanup = LocalTime.now(clock).plus(this.cleanupInterval);