GH-529: Add lz4 and docs for zstd compression
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/529 - Also log exception when getting partition information
This commit is contained in:
@@ -317,6 +317,13 @@ If a topic already exists with a smaller partition count and `autoAddPartitions`
|
||||
If a topic already exists with a smaller partition count and `autoAddPartitions` is enabled, new partitions are added.
|
||||
If a topic already exists with a larger number of partitions than the maximum of (`minPartitionCount` or `partitionCount`), the existing partition count is used.
|
||||
|
||||
compression::
|
||||
Set the `compression.type` producer property.
|
||||
Supported values are `none`, `gzip`, `snappy` and `lz4`.
|
||||
If you override the `kafka-clients` jar to 2.1.0 (or later), as discussed in the https://docs.spring.io/spring-kafka/docs/2.2.x/reference/html/deps-for-21x.html[Spring for Apache Kafka documentation], and wish to use `zstd` compression, use `spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration.compression.type=zstd`.
|
||||
+
|
||||
Default: `none`.
|
||||
|
||||
==== Usage examples
|
||||
|
||||
In this section, we show the use of the preceding properties for specific scenarios.
|
||||
|
||||
@@ -117,17 +117,32 @@ public class KafkaProducerProperties {
|
||||
* Enumeration for compression types.
|
||||
*/
|
||||
public enum CompressionType {
|
||||
|
||||
/**
|
||||
* No compression.
|
||||
*/
|
||||
none,
|
||||
|
||||
/**
|
||||
* gzip based compression.
|
||||
*/
|
||||
gzip,
|
||||
|
||||
/**
|
||||
* snappy based compression.
|
||||
*/
|
||||
snappy
|
||||
snappy,
|
||||
|
||||
/**
|
||||
* lz4 compression
|
||||
*/
|
||||
lz4,
|
||||
|
||||
// /** // TODO: uncomment and fix docs when kafka-clients 2.1.0 or newer is the default
|
||||
// * zstd compression
|
||||
// */
|
||||
// zstd
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -414,6 +414,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
|
||||
if (ex instanceof UnknownTopicOrPartitionException) {
|
||||
throw ex;
|
||||
}
|
||||
this.logger.error("Failed to obtain partition information", ex);
|
||||
}
|
||||
if (CollectionUtils.isEmpty(partitions)) {
|
||||
final AdminClient adminClient = AdminClient.create(this.adminClientProperties);
|
||||
|
||||
Reference in New Issue
Block a user