Skip to content

KAFKA-7632: Support Compression Level #10826

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dongjinleekr wants to merge 9 commits into from
Closed

KAFKA-7632: Support Compression Level #10826

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
6c10e27
Add CompressionConfig and separate compression configs per codec: Non…
dongjinleekr May 27, 2021
d493d41
Make GzipConfig to support compression level
dongjinleekr Oct 2, 2021
db7d404
Make LZ4Config to support compression level
dongjinleekr Oct 2, 2021
29dbaf2
Make ZstdConfig to support compression level
dongjinleekr Oct 2, 2021
89c1182
Add compression.level configuration
dongjinleekr Oct 2, 2021
b334052
Rename class name: KafkaLZ4BlockOutputStream → Lz4OutputStream
dongjinleekr Jul 8, 2022
8ef745b
Change method names: getType → type. setLevel → level.
dongjinleekr Jul 8, 2022
c951110
Trivial improvements
dongjinleekr Jul 8, 2022
2c96abf
Remove unused constructors in Lz4OutputStream
dongjinleekr Jul 8, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions checkstyle/import-control.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,9 @@
<allow pkg="net.jpountz.xxhash" />
<allow pkg="org.apache.kafka.common.compress" />
<allow pkg="org.xerial.snappy" />
<allow class="org.apache.kafka.common.record.CompressionConfig" exact-match="true" />
<allow class="org.apache.kafka.common.record.CompressionType" exact-match="true" />
<allow class="org.apache.kafka.common.record.RecordBatch.MAGIC_VALUE_V0" exact-match="true" />
</subpackage>

<subpackage name="message">
Expand Down
2 changes: 1 addition & 1 deletion checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@
files="(Errors|SaslAuthenticatorTest|AgentTest|CoordinatorTest).java"/>

<suppress checks="BooleanExpressionComplexity"
files="(Utils|Topic|KafkaLZ4BlockOutputStream|AclData|JoinGroupRequest).java"/>
files="(Utils|Topic|Lz4OutputStream|AclData|JoinGroupRequest).java"/>

<suppress checks="CyclomaticComplexity"
files="(ConsumerCoordinator|Fetcher|KafkaProducer|ConfigDef|KerberosLogin|AbstractRequest|AbstractResponse|Selector|SslFactory|SslTransportLayer|SaslClientAuthenticator|SaslClientCallbackHandler|SaslServerAuthenticator|AbstractCoordinator|TransactionManager|AbstractStickyAssignor|DefaultSslEngineFactory|Authorizer|RecordAccumulator|MemoryRecords|FetchSessionHandler).java"/>
Expand Down
12 changes: 6 additions & 6 deletions clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@
import org.apache.kafka.common.network.ChannelBuilder;
import org.apache.kafka.common.network.Selector;
import org.apache.kafka.common.record.AbstractRecords;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.record.CompressionConfig;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.requests.JoinGroupRequest;
import org.apache.kafka.common.serialization.Serializer;
Expand Down Expand Up @@ -248,7 +248,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
private final RecordAccumulator accumulator;
private final Sender sender;
private final Thread ioThread;
private final CompressionType compressionType;
private final CompressionConfig compressionConfig;
private final Sensor errors;
private final Time time;
private final Serializer<K> keySerializer;
Expand Down Expand Up @@ -411,7 +411,7 @@ private void warnIfPartitionerDeprecated() {
valueSerializer, interceptorList, reporters);
this.maxRequestSize = config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG);
this.totalMemorySize = config.getLong(ProducerConfig.BUFFER_MEMORY_CONFIG);
this.compressionType = CompressionType.forName(config.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
this.compressionConfig = config.getCompressionConfig();

this.maxBlockTimeMs = config.getLong(ProducerConfig.MAX_BLOCK_MS_CONFIG);
int deliveryTimeoutMs = configureDeliveryTimeout(config, log);
Expand All @@ -427,7 +427,7 @@ private void warnIfPartitionerDeprecated() {
);
this.accumulator = new RecordAccumulator(logContext,
config.getInt(ProducerConfig.BATCH_SIZE_CONFIG),
this.compressionType,
this.compressionConfig,
lingerMs(config),
retryBackoffMs,
deliveryTimeoutMs,
Expand Down Expand Up @@ -495,7 +495,7 @@ private void warnIfPartitionerDeprecated() {
this.interceptors = interceptors;
this.maxRequestSize = config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG);
this.totalMemorySize = config.getLong(ProducerConfig.BUFFER_MEMORY_CONFIG);
this.compressionType = CompressionType.forName(config.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
this.compressionConfig = config.getCompressionConfig();
this.maxBlockTimeMs = config.getLong(ProducerConfig.MAX_BLOCK_MS_CONFIG);
this.partitionerIgnoreKeys = config.getBoolean(ProducerConfig.PARTITIONER_IGNORE_KEYS_CONFIG);
this.apiVersions = new ApiVersions();
Expand Down Expand Up @@ -1015,7 +1015,7 @@ private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback call
Header[] headers = record.headers().toArray();

int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(apiVersions.maxUsableProduceMagic(),
compressionType, serializedKey, serializedValue, headers);
compressionConfig.type(), serializedKey, serializedValue, headers);
ensureValidRecordSize(serializedSize);
long timestamp = record.timestamp() == null ? nowMs : record.timestamp();

Expand Down
47 changes: 47 additions & 0 deletions clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.SecurityConfig;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.record.CompressionConfig;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.serialization.Serializer;
Expand Down Expand Up @@ -216,6 +217,12 @@ public class ProducerConfig extends AbstractConfig {
+ " values are <code>none</code>, <code>gzip</code>, <code>snappy</code>, <code>lz4</code>, or <code>zstd</code>. "
+ "Compression is of full batches of data, so the efficacy of batching will also impact the compression ratio (more batching means better compression).";

/** <code>compression.level</code> */
public static final String COMPRESSION_LEVEL_CONFIG = "compression.level";
private static final String COMPRESSION_LEVEL_DOC = "The compression level for all data generated by the producer. The default level and valid value is up to "
+ "compression.type. (<code>none</code>, <code>snappy</code>: not available. <code>gzip</code>: 1~9. <code>lz4</code>: 1~17. "

@mimaison mimaison Apr 14, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we include the defaults too?

@ijuma ijuma Apr 14, 2022 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the reason why @dongjinleekr didn't include those is that they may change at the compression library level.

+ "<code>zstd</code>: -131072~22.";

/** <code>metrics.sample.window.ms</code> */
public static final String METRICS_SAMPLE_WINDOW_MS_CONFIG = CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_CONFIG;

Expand Down Expand Up @@ -350,6 +357,7 @@ public class ProducerConfig extends AbstractConfig {
Importance.LOW,
ACKS_DOC)
.define(COMPRESSION_TYPE_CONFIG, Type.STRING, CompressionType.NONE.name, in(Utils.enumOptions(CompressionType.class)), Importance.HIGH, COMPRESSION_TYPE_DOC)
.define(COMPRESSION_LEVEL_CONFIG, Type.STRING, "", Importance.MEDIUM, COMPRESSION_LEVEL_DOC)

@junrao junrao Apr 13, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The KIP says the default value is null.

@dongjinleekr dongjinleekr Jul 8, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you kindly review the discussion here?

.define(BATCH_SIZE_CONFIG, Type.INT, 16384, atLeast(0), Importance.MEDIUM, BATCH_SIZE_DOC)
.define(PARTITIONER_ADPATIVE_PARTITIONING_ENABLE_CONFIG, Type.BOOLEAN, true, Importance.LOW, PARTITIONER_ADPATIVE_PARTITIONING_ENABLE_DOC)
.define(PARTITIONER_AVAILABILITY_TIMEOUT_MS_CONFIG, Type.LONG, 0, atLeast(0), Importance.LOW, PARTITIONER_AVAILABILITY_TIMEOUT_MS_DOC)
Expand Down Expand Up @@ -583,6 +591,45 @@ public ProducerConfig(Map<String, Object> props) {
super(CONFIG, props);
}

CompressionConfig getCompressionConfig(CompressionType compressionType) {
if (getString(ProducerConfig.COMPRESSION_LEVEL_CONFIG).isEmpty()) {

@mimaison mimaison Apr 14, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since only 3 cases use the compression level, what about moving the if/else inside each case block?

Suggested change
if (getString(ProducerConfig.COMPRESSION_LEVEL_CONFIG).isEmpty()) {
public CompressionConfig getCompressionConfig(CompressionType compressionType) {
String compressionLevel = getString(ProducerConfig.COMPRESSION_LEVEL_CONFIG);
switch (compressionType) {
case NONE:
return CompressionConfig.NONE;
case GZIP:
if (compressionLevel.isEmpty())
return CompressionConfig.gzip().build();
else
return CompressionConfig.gzip().setLevel(Integer.parseInt(compressionLevel)).build();
case SNAPPY:
return CompressionConfig.snappy().build();
case LZ4:
if (compressionLevel.isEmpty())
return CompressionConfig.lz4().build();
else
return CompressionConfig.lz4().setLevel(Integer.parseInt(compressionLevel)).build();
case ZSTD:
if (compressionLevel.isEmpty())
return CompressionConfig.zstd().build();
else
return CompressionConfig.zstd().setLevel(Integer.parseInt(compressionLevel)).build();
default:
throw new IllegalArgumentException("Unknown compression type: " + compressionType.name);
}
}

@ijuma ijuma Apr 14, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why don't we push some of these things to the enum itself? CompressionType.config(String level) or something like that.

@dongjinleekr dongjinleekr Jul 8, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The reason why I did not choose a simple approach like CompressionType.config(String level) (as @ijuma pointed out) here is simple: To make room for supporting per-codec configuration in the future, like KIP-780.

The reason why I chose to use compression.level without default value (as @mimaison pointed out) is also simple: at the point of when KIP-390 is passed, the proposal included global configuration option (i.e., compression.level). But, as I work on KIP-780 and running an in-house preview, I found that per-codec configuration (i.e., compression.{gzip|lz4|zstd}.level) is much more preferable for the following reasons:

1. Easy to switch the compression type

For example, {compression.type=zstd,compression.level=10} works but {compression.type=gzip,compression.level=10} does not, since gzip only supports the level of 1 to 9. In other words, compression.type is so error-prone when the user tries to switch the codec. (@junrao indicated a similar problem: #1 #2)

If we limit the scope of per-codec options like compression.gzip.level, we can easily switch the whole options by changing compression.type only. Moreover, this approach is consistent with additional compression options like compression.gzip.buffer.

2. Can provide a default level.

As @ijuma pointed out, we can not provide a default compression level with compression.level; but with compression.{gzip|lz4|zstd}.level, we can.

For these reasons, I run the overall benchmarks in here with per-config options (see 'Final Notes' section.) If the reviewers prefer this scheme, I will happily change it.

@mimaison mimaison Aug 8, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @dongjinleekr for the explanations. My suggested change was logically identical, it's just a bit shorter.

switch (compressionType) {
case NONE:
return CompressionConfig.NONE;
case GZIP:
return CompressionConfig.gzip().build();
case SNAPPY:
return CompressionConfig.snappy().build();
case LZ4:
return CompressionConfig.lz4().build();
case ZSTD:
return CompressionConfig.zstd().build();
default:
throw new IllegalArgumentException("Unknown compression type: " + compressionType.name);
}
} else {
switch (compressionType) {
case NONE:
return CompressionConfig.NONE;
case GZIP:
return CompressionConfig.gzip().level(Integer.parseInt(getString(ProducerConfig.COMPRESSION_LEVEL_CONFIG))).build();
case SNAPPY:
return CompressionConfig.snappy().build();
case LZ4:
return CompressionConfig.lz4().level(Integer.parseInt(getString(ProducerConfig.COMPRESSION_LEVEL_CONFIG))).build();
case ZSTD:
return CompressionConfig.zstd().level(Integer.parseInt(getString(ProducerConfig.COMPRESSION_LEVEL_CONFIG))).build();
default:
throw new IllegalArgumentException("Unknown compression type: " + compressionType.name);
}
}
}

public CompressionConfig getCompressionConfig() {
CompressionType compressionType = CompressionType.forName(getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
return getCompressionConfig(compressionType);
}

ProducerConfig(Map<?, ?> props, boolean doLog) {
super(CONFIG, props, doLog);
}
Expand Down
14 changes: 7 additions & 7 deletions clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ public ProducerBatch(TopicPartition tp, MemoryRecordsBuilder recordsBuilder, lon
this.retry = false;
this.isSplitBatch = isSplitBatch;
float compressionRatioEstimation = CompressionRatioEstimator.estimation(topicPartition.topic(),
recordsBuilder.compressionType());
recordsBuilder.compressionConfig().type());
recordsBuilder.setEstimatedCompressionRatio(compressionRatioEstimation);
}

Expand All @@ -108,7 +108,7 @@ public FutureRecordMetadata tryAppend(long timestamp, byte[] key, byte[] value,
} else {
this.recordsBuilder.append(timestamp, key, value, headers);
this.maxRecordSize = Math.max(this.maxRecordSize, AbstractRecords.estimateSizeInBytesUpperBound(magic(),
recordsBuilder.compressionType(), key, value, headers));
recordsBuilder.compressionConfig().type(), key, value, headers));
this.lastAppendTime = now;
FutureRecordMetadata future = new FutureRecordMetadata(this.produceFuture, this.recordCount,
timestamp,
Expand All @@ -134,7 +134,7 @@ private boolean tryAppendForSplit(long timestamp, ByteBuffer key, ByteBuffer val
// No need to get the CRC.
this.recordsBuilder.append(timestamp, key, value, headers);
this.maxRecordSize = Math.max(this.maxRecordSize, AbstractRecords.estimateSizeInBytesUpperBound(magic(),
recordsBuilder.compressionType(), key, value, headers));
recordsBuilder.compressionConfig().type(), key, value, headers));
FutureRecordMetadata future = new FutureRecordMetadata(this.produceFuture, this.recordCount,
timestamp,
key == null ? -1 : key.remaining(),
Expand Down Expand Up @@ -339,19 +339,19 @@ public Deque<ProducerBatch> split(int splitBatchSize) {

private ProducerBatch createBatchOffAccumulatorForRecord(Record record, int batchSize) {
int initialSize = Math.max(AbstractRecords.estimateSizeInBytesUpperBound(magic(),
recordsBuilder.compressionType(), record.key(), record.value(), record.headers()), batchSize);
recordsBuilder.compressionConfig().type(), record.key(), record.value(), record.headers()), batchSize);
ByteBuffer buffer = ByteBuffer.allocate(initialSize);

// Note that we intentionally do not set producer state (producerId, epoch, sequence, and isTransactional)
// for the newly created batch. This will be set when the batch is dequeued for sending (which is consistent
// with how normal batches are handled).
MemoryRecordsBuilder builder = MemoryRecords.builder(buffer, magic(), recordsBuilder.compressionType(),
MemoryRecordsBuilder builder = MemoryRecords.builder(buffer, magic(), recordsBuilder.compressionConfig(),
TimestampType.CREATE_TIME, 0L);
return new ProducerBatch(topicPartition, builder, this.createdMs, true);
}

public boolean isCompressed() {
return recordsBuilder.compressionType() != CompressionType.NONE;
return recordsBuilder.compressionConfig().type() != CompressionType.NONE;
}

/**
Expand Down Expand Up @@ -453,7 +453,7 @@ public void close() {
recordsBuilder.close();
if (!recordsBuilder.isControlBatch()) {
CompressionRatioEstimator.updateEstimation(topicPartition.topic(),
recordsBuilder.compressionType(),
recordsBuilder.compressionConfig().type(),
(float) recordsBuilder.compressionRatio());
}
reopened = false;
Expand Down
20 changes: 10 additions & 10 deletions clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.record.AbstractRecords;
import org.apache.kafka.common.record.CompressionRatioEstimator;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.record.CompressionConfig;
import org.apache.kafka.common.record.MemoryRecords;
import org.apache.kafka.common.record.MemoryRecordsBuilder;
import org.apache.kafka.common.record.Record;
Expand All @@ -70,7 +70,7 @@ public class RecordAccumulator {
private final AtomicInteger flushesInProgress;
private final AtomicInteger appendsInProgress;
private final int batchSize;
private final CompressionType compression;
private final CompressionConfig compression;
private final int lingerMs;
private final long retryBackoffMs;
private final int deliveryTimeoutMs;
Expand All @@ -93,7 +93,7 @@ public class RecordAccumulator {
*
* @param logContext The log context used for logging
* @param batchSize The size to use when allocating {@link MemoryRecords} instances
* @param compression The compression codec for the records
* @param compressionConfig The compression configuration for the records
* @param lingerMs An artificial delay time to add before declaring a records instance that isn't full ready for
* sending. This allows time for more records to arrive. Setting a non-zero lingerMs will trade off some
* latency for potentially better throughput due to more batching (and hence fewer, larger requests).
Expand All @@ -111,7 +111,7 @@ public class RecordAccumulator {
*/
public RecordAccumulator(LogContext logContext,
int batchSize,
CompressionType compression,
CompressionConfig compressionConfig,
int lingerMs,
long retryBackoffMs,
int deliveryTimeoutMs,
Expand All @@ -128,7 +128,7 @@ public RecordAccumulator(LogContext logContext,
this.flushesInProgress = new AtomicInteger(0);
this.appendsInProgress = new AtomicInteger(0);
this.batchSize = batchSize;
this.compression = compression;
this.compression = compressionConfig;
this.lingerMs = lingerMs;
this.retryBackoffMs = retryBackoffMs;
this.deliveryTimeoutMs = deliveryTimeoutMs;
Expand All @@ -149,7 +149,7 @@ public RecordAccumulator(LogContext logContext,
*
* @param logContext The log context used for logging
* @param batchSize The size to use when allocating {@link MemoryRecords} instances
* @param compression The compression codec for the records
* @param compressionConfig The compression configuration for the records
* @param lingerMs An artificial delay time to add before declaring a records instance that isn't full ready for
* sending. This allows time for more records to arrive. Setting a non-zero lingerMs will trade off some
* latency for potentially better throughput due to more batching (and hence fewer, larger requests).
Expand All @@ -166,7 +166,7 @@ public RecordAccumulator(LogContext logContext,
*/
public RecordAccumulator(LogContext logContext,
int batchSize,
CompressionType compression,
CompressionConfig compressionConfig,
int lingerMs,
long retryBackoffMs,
int deliveryTimeoutMs,
Expand All @@ -178,7 +178,7 @@ public RecordAccumulator(LogContext logContext,
BufferPool bufferPool) {
this(logContext,
batchSize,
compression,
compressionConfig,
lingerMs,
retryBackoffMs,
deliveryTimeoutMs,
Expand Down Expand Up @@ -295,7 +295,7 @@ public RecordAppendResult append(String topic,

if (buffer == null) {
byte maxUsableMagic = apiVersions.maxUsableProduceMagic();
int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(maxUsableMagic, compression, key, value, headers));
int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(maxUsableMagic, compression.type(), key, value, headers));
log.trace("Allocating a new {} byte message buffer for topic {} partition {} with remaining timeout {}ms", size, topic, partition, maxTimeToBlock);
buffer = free.allocate(size, maxTimeToBlock);
}
Expand Down Expand Up @@ -470,7 +470,7 @@ public int splitAndReenqueue(ProducerBatch bigBatch) {
// Reset the estimated compression ratio to the initial value or the big batch compression ratio, whichever
// is bigger. There are several different ways to do the reset. We chose the most conservative one to ensure
// the split doesn't happen too often.
CompressionRatioEstimator.setEstimation(bigBatch.topicPartition.topic(), compression,
CompressionRatioEstimator.setEstimation(bigBatch.topicPartition.topic(), compression.type(),
Math.max(1.0f, (float) bigBatch.compressionRatio()));
Deque<ProducerBatch> dq = bigBatch.split(this.batchSize);
int numSplitBatches = dq.size();
Expand Down
Loading