Parameter to only show the latest sate in topics with log compaction

[1nval1d]

Hi,

I started using a lot of log compaction and am missing a feature from kafkactl.
It would be quite useful if the tool only showed the latest message for a key and excluded removed keys entirely.
So if I had a compacted topic with messages for 3 keys, one just one message with offset 1, one a start (offset 2) and an update (offset 3) and the last one a start (offset 4) and a tombstone (offset 5), Id like to only get messages 1 and 3.

It would be awesome if that even worked with the tail command so the tool delivered the amount of messaged specified in the parameter after the filtering for valid keys, but to my knowledge that's not really possible or is it? You'd need to consume a topic backwards with decreasing offsets for that.

As this is not expected behaviour, its probably a bad idea to make it the default for topics with compaction but a parameter to enable it would be great.

I have no clue of Go, otherwise Id create a PR for it but Id really appreciate it, if that was added.
If I can be of useful help, let me know.

Thank you

[d-rk]

Hey @1nval1d

to summarize your example, here are the commands to rebuild your scenario:

$ kafkactl create topic compacted-test --config cleanup.policy=compact
$ kafkactl produce compacted-test --key=key-1 --value=value-1
$ kafkactl produce compacted-test --key=key-2 --value=value-start
$ kafkactl produce compacted-test --key=key-2 --value=value-update
$ kafkactl produce compacted-test --key=key-3 --value=value-start
$ kafkactl produce compacted-test --key=key-3 --null-value

$ kafkactl consume compacted-test --from-beginning --print-keys -o yaml --exit
partition: 0
offset: 0
key: key-1
value: value-1

partition: 0
offset: 1
key: key-2
value: value-start

partition: 0
offset: 2
key: key-2
value: value-update

partition: 0
offset: 3
key: key-3
value: value-start

partition: 0
offset: 4
key: key-3
value: null

To see what would be the expected output we could trigger a log compaction after 10s with this:

$ kafkactl alter topic compacted-test --config max.compaction.lag.ms=10000 --config min.cleanable.dirty.ratio=0.0 --config segment.ms=10000 --config delete.retention.ms=10000
# we need to write to the topic to trigger the compaction
$ kafkactl produce compacted-test --key=key-4 --value=value-4

$ kafkactl consume compacted-test --from-beginning --print-keys -o yaml --exit
partition: 0
offset: 0
key: key-1
value: value-1

partition: 0
offset: 2
key: key-2
value: value-update

partition: 0
offset: 4
key: key-3
value: null

partition: 0
offset: 5
key: key-4
value: value-4

We will probably not have the time to implement such a feature in the near future. But if someone is willing to provide a PR I'll be happy to help.