Сжатие Kafka Log возвращает две записи с одинаковым ключом

Question

У меня странное поведение при сжатии журналов в kafka.Я создал тему со следующей конфигурацией:

kafka-topics --zookeeper ... \
--create --topic myTopic \
--partitions 12 \
--replication-factor 3 \
--config "min.insync.replicas=2" \
--config "cleanup.policy=compact" \
--config "delete.retention.ms=100" \
--config "retention.bytes=-1" \
--config "segment.ms=100" \
--config "min.cleanable.dirty.ratio=0.000001" \
--config "min.compaction.lag.ms=10"

Я отправляю сообщения с тем же ключом в тему, и когда запускается компиляция, он возвращает два последних сообщения этой темы.

Пример:

Writting message with key="1" and value="A" into topic "myTopic"
Writting message with key="1" and value="B" into topic "myTopic"
Writting message with key="1" and value="C" into topic "myTopic"

СЖАТИЕ

печать 'myTopic' с начала

{"ROWTIME": 1549444994905, "ROWKEY": "1", "value =B "} {" ROWTIME ": 1549444994905," ROWKEY ":" 1 "," value = C "}

Writting message with key="1" and value="D" into topic "myTopic"

COMPACTION

вывод 'myTopic' с начала

{"ROWTIME": 1549444994905, "ROWKEY": "1", "value = C"} {"ROWTIME": 1549444994905, "ROWKEY": "1", "value = D"}

То же самое с сегментом .ms = 60000

Есть идеи?

Спасибо !!

Answer 1

До уплотнения Кафка определил нижнюю позицию смещения, которая не может принимать участие в уплотнении (firstUncleanableDirtyOffset).

Эта позиция рассчитывается на основе:

• Первое нестабильное смещение
• Активное смещение сегмента
• min.compaction.lag.ms

В вашем случае min.compaction.lag.ms очень низкий, поэтому самая низкая позиция смещения (которая не может участвовать в уплотнении) берется из активного сегмента.Из-за этого только одно сообщение может принимать участие в сжатии (например, ключ = 1, значение = C), поэтому ничего не нужно делать.

Если вы создаете дополнительное сообщение с другим ключом, оно должно сжать ваши сообщения для key=1.

Примечание: Вам также необходимо знать свойство segment.bytes, которое определяет размер сегмента.Если сообщения небольшие по сравнению с segment.bytes, они могут находиться в активном сегменте и не будут уплотнены.