KAFKA-19616: Add compression type and level support to LogCompactionTester

[yunchipang]

issue: KAFKA-19616

why: validate log compaction works correctly with compressed data.
what: adds compression config options to LogCompactionTester tool
and extends test coverage to validate log compaction with different
compression types and levels.

Reviewers: TengYao Chi [email protected], Chia-Ping Tsai
[email protected]

[yunchipang]

e2e test result

SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/kafka-dev/core/build/dependant-testlibs/log4j-slf4j-impl-2.24.3.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/kafka-dev/tools/build/dependant-libs-2.13.16/log4j-slf4j-impl-2.24.3.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
Producing 1000000 messages..to topics log-cleaner-test-1163149033400908362-0
Logging produce requests to /tmp/kafka-log-cleaner-produced-2674895871657186239.txt
Sleeping for 20seconds...
Consuming messages...
Logging consumed messages to /tmp/kafka-log-cleaner-consumed-6636562293354175110.txt
1000000 rows of data produced, 120154 rows of data consumed (88.0% reduction).
De-duplicating and validating output files...
Validated 90004 values, 0 mismatches.
Data verification is completed

[chia7712]

@yunchipang Could you please fix the conflicts?

[frankvicky]

@yunchipang: Please apply spotless.

[chia7712]

@yunchipang thanks for your patch!

[chia7712]

@yunchipang thanks for the updates!

[yunchipang]

e2e test result

SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/kafka-dev/core/build/dependant-testlibs/log4j-slf4j-impl-2.25.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/kafka-dev/tools/build/dependant-libs-2.13.16/log4j-slf4j-impl-2.25.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
Producing 1000000 messages..to topics log-cleaner-test-1932846938360647426-0
Logging produce requests to /tmp/kafka-log-cleaner-produced-4893708329550575535.txt
Sleeping for 20seconds...
Consuming messages...
Logging consumed messages to /tmp/kafka-log-cleaner-consumed-13046421783240844934.txt
1000000 rows of data produced, 120164 rows of data consumed (88.0% reduction).
De-duplicating and validating output files...
Validated 89942 values, 0 mismatches.
Data verification is completed

[yunchipang]

running TC_PATHS="tests/kafkatest/tests/tools/log_compaction_test.py" /bin/bash tests/docker/run_tests.sh

================================================================================
SESSION REPORT (ALL TESTS)
ducktape version: 0.12.0
session_id:       2025-08-27--001
run time:         7 minutes 46.138 seconds
tests run:        8
passed:           8
flaky:            0
failed:           0
ignored:          0
================================================================================
test_id:    kafkatest.tests.tools.log_compaction_test.LogCompactionTest.test_log_compaction.metadata_quorum=ISOLATED_KRAFT.compression_config=
status:     PASS
run time:   59.595 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.tools.log_compaction_test.LogCompactionTest.test_log_compaction.metadata_quorum=ISOLATED_KRAFT.compression_config=.type.gzip.level.1
status:     PASS
run time:   57.970 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.tools.log_compaction_test.LogCompactionTest.test_log_compaction.metadata_quorum=ISOLATED_KRAFT.compression_config=.type.gzip.level.9
status:     PASS
run time:   57.639 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.tools.log_compaction_test.LogCompactionTest.test_log_compaction.metadata_quorum=ISOLATED_KRAFT.compression_config=.type.lz4.level.1
status:     PASS
run time:   57.287 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.tools.log_compaction_test.LogCompactionTest.test_log_compaction.metadata_quorum=ISOLATED_KRAFT.compression_config=.type.lz4.level.10
status:     PASS
run time:   57.778 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.tools.log_compaction_test.LogCompactionTest.test_log_compaction.metadata_quorum=ISOLATED_KRAFT.compression_config=.type.snappy
status:     PASS
run time:   58.785 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.tools.log_compaction_test.LogCompactionTest.test_log_compaction.metadata_quorum=ISOLATED_KRAFT.compression_config=.type.zstd.level.1
status:     PASS
run time:   57.755 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.tools.log_compaction_test.LogCompactionTest.test_log_compaction.metadata_quorum=ISOLATED_KRAFT.compression_config=.type.zstd.level.10
status:     PASS
run time:   58.170 seconds
--------------------------------------------------------------------------------

[chia7712]

@yunchipang thanks for your updates

[chia7712]

the failed test is already fixed by #20419