Не могу заставить работать уплотнение в Apache Kafka ⇐ JAVA

[Anonymous]

У меня есть проблемы с компактом политики очистки Kafka. Но я не могу заставить его работать должным образом.
шаг к воспроизведению: < /p>

• Создать контейнер Docker < /li>
  < /ol>
  docker run -d \
  --name kafka \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -p 9092:9092 \
  -p 9999:9999 \
  -e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
  -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
  -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
  -e KAFKA_NUM_PARTITIONS=3 \
  -e KAFKA_LOG_CLEANER_ENABLE=true \
  -e KAFKA_LOG_CLEANER_THREADS=2 \
  -e KAFKA_LOG_CLEANER_DEDUPE_BUFFER_SIZE=134217728 \
  -e KAFKA_LOG_CLEANER_IO_BUFFER_SIZE=524288 \
  -e KAFKA_JMX_PORT=9999 \
  apache/kafka:3.7.2
  < /code>
  
  Создать тему < /li>
  < /ol>
  docker exec -it kafka bash
  < /code>
  cd /opt/kafka/bin/
  < /code>
  ./kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --topic compact-topic1 \
  --partitions 4 \
  --replication-factor 1 \
  --config cleanup.policy=compact \
  --config segment.bytes=100000 \
  --config min.cleanable.dirty.ratio=0.01 \
  --config delete.retention.ms=10000 \
  --config min.compaction.lag.ms=5000 \
  --config max.compaction.lag.ms=30000 \
  --config segment.ms=60000
  < /code>
  
  send events (SpringBoot 3.4.5)

package com.example.kafka_compact_playground;

import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

@RequiredArgsConstructor
@Component
@Slf4j
public class KafkaGeneratorService {
private static final String TOPIC = "compact-topic1";
private final KafkaTemplate kafkaTemplate;

public void sendMessage(String key, String message) {
// For compact topics, always use a key
kafkaTemplate.send(TOPIC, key, message)

;
}

@PostConstruct
void init(){
int k = 0;
for(int j = 0; j < 500000; j++) {
for (int i = 0; i < 10; i++) { // More keys
String key = "key" + i;
String message = "message" + i + "_version_" + j + " data: " + LocalDateTime.now();
sendMessage(key, message);
k++;

if (k % 1000 == 0) {
log.info("Batch {}: Sent message with key: {}", k, key);
try {
Thread.sleep(10);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
}

// Add small delay to allow segments to close

}
}
}
< /code>
Waiting...
Check with offset explorer - 5000000 events
What I'm expecting - a little bit less events.
What I'm doing wrong?

Подробнее здесь: https://stackoverflow.com/questions/796 ... ache-kafka