Skip to content

Commit

Permalink
[improve][fn] Support configure compression type (#19470)
Browse files Browse the repository at this point in the history
  • Loading branch information
@jiangpengcheng
jiangpengcheng authored Mar 17, 2023
1 parent ad43c63 commit e0098ee
Show file tree
Hide file tree
Showing 13 changed files with 578 additions and 315 deletions.
2 changes: 2 additions & 0 deletions pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/functions/ProducerConfig.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import org.apache.pulsar.client.api.CompressionType;

/**
* Configuration of the producer inside the function.
Expand All @@ -38,4 +39,5 @@ public class ProducerConfig {
private Boolean useThreadLocalProducers;
private CryptoConfig cryptoConfig;
private String batchBuilder;
private CompressionType compressionType;
}
654 changes: 377 additions & 277 deletions pulsar-function-go/pb/Function.pb.go
View file

Large diffs are not rendered by default.

14 changes: 13 additions & 1 deletion pulsar-function-go/pf/instance.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -225,7 +225,19 @@ func (gi *goInstance) getProducer(topicName string) (pulsar.Producer, error) {

batchBuilderType := pulsar.DefaultBatchBuilder

compressionType := pulsar.LZ4
if gi.context.instanceConf.funcDetails.Sink.ProducerSpec != nil {
switch gi.context.instanceConf.funcDetails.Sink.ProducerSpec.CompressionType {
case pb.CompressionType_NONE:
compressionType = pulsar.NoCompression
case pb.CompressionType_ZLIB:
compressionType = pulsar.ZLib
case pb.CompressionType_ZSTD:
compressionType = pulsar.ZSTD
default:
compressionType = pulsar.LZ4 // go doesn't support SNAPPY yet
}

batchBuilder := gi.context.instanceConf.funcDetails.Sink.ProducerSpec.BatchBuilder
if batchBuilder != "" {
if batchBuilder == "KEY_BASED" {
Expand All @@ -237,7 +249,7 @@ func (gi *goInstance) getProducer(topicName string) (pulsar.Producer, error) {
producer, err := gi.client.CreateProducer(pulsar.ProducerOptions{
Topic: topicName,
Properties: properties,
CompressionType: pulsar.LZ4,
CompressionType: compressionType,
BatchingMaxPublishDelay: time.Millisecond * 10,
BatcherBuilderType: batchBuilderType,
SendTimeout: 0,
Expand Down
4 changes: 3 additions & 1 deletion ...ons/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -904,7 +904,9 @@ private void setupOutput(ContextImpl contextImpl) throws Exception {
.maxPendingMessagesAcrossPartitions(conf.getMaxPendingMessagesAcrossPartitions())
.batchBuilder(conf.getBatchBuilder())
.useThreadLocalProducers(conf.getUseThreadLocalProducers())
.cryptoConfig(CryptoUtils.convertFromSpec(conf.getCryptoSpec()));
.cryptoConfig(CryptoUtils.convertFromSpec(conf.getCryptoSpec()))
.compressionType(FunctionCommon.convertFromFunctionDetailsCompressionType(
conf.getCompressionType()));
pulsarSinkConfig.setProducerConfig(builder.build());
}

Expand Down
6 changes: 5 additions & 1 deletion pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,6 @@ public Producer<T> createProducer(PulsarClient client, String topic, String prod
.blockIfQueueFull(true)
.enableBatching(true)
.batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
.compressionType(CompressionType.LZ4)
.hashingScheme(HashingScheme.Murmur3_32Hash) //
.messageRoutingMode(MessageRoutingMode.CustomPartition)
.messageRouter(FunctionResultRouter.of())
Expand All @@ -121,6 +120,11 @@ public Producer<T> createProducer(PulsarClient client, String topic, String prod
}
if (pulsarSinkConfig.getProducerConfig() != null) {
ProducerConfig producerConfig = pulsarSinkConfig.getProducerConfig();
if (producerConfig.getCompressionType() != null) {
builder.compressionType(producerConfig.getCompressionType());
} else {
builder.compressionType(CompressionType.LZ4);
}
if (producerConfig.getMaxPendingMessages() != 0) {
builder.maxPendingMessages(producerConfig.getMaxPendingMessages());
}
Expand Down
131 changes: 97 additions & 34 deletions pulsar-functions/instance/src/main/python/Function_pb2.py
View file

Large diffs are not rendered by default.

13 changes: 12 additions & 1 deletion pulsar-functions/instance/src/main/python/python_instance.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -352,14 +352,25 @@ def setup_producer(self):
if crypto_key_reader is not None:
encryption_key = self.instance_config.function_details.sink.producerSpec.cryptoSpec.producerEncryptionKeyName[0]

compression_type = pulsar.CompressionType.LZ4
if self.instance_config.function_details.sink.producerSpec.compressionType is not None:
if self.instance_config.function_details.sink.producerSpec.compressionType == Function_pb2.CompressionType.Value("NONE"):
compression_type = pulsar.CompressionType.NONE
elif self.instance_config.function_details.sink.producerSpec.compressionType == Function_pb2.CompressionType.Value("ZLIB"):
compression_type = pulsar.CompressionType.ZLib
elif self.instance_config.function_details.sink.producerSpec.compressionType == Function_pb2.CompressionType.Value("ZSTD"):
compression_type = pulsar.CompressionType.ZSTD
elif self.instance_config.function_details.sink.producerSpec.compressionType == Function_pb2.CompressionType.Value("SNAPPY"):
compression_type = pulsar.CompressionType.SNAPPY

self.producer = self.pulsar_client.create_producer(
str(self.instance_config.function_details.sink.topic),
schema=self.output_schema,
block_if_queue_full=True,
batching_enabled=True,
batching_type=batch_type,
batching_max_publish_delay_ms=10,
compression_type=pulsar.CompressionType.LZ4,
compression_type=compression_type,
# set send timeout to be infinity to prevent potential deadlock with consumer
# that might happen when consumer is blocked due to unacked messages
send_timeout_millis=0,
Expand Down
9 changes: 9 additions & 0 deletions pulsar-functions/proto/src/main/proto/Function.proto
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,14 @@ enum SubscriptionPosition {
EARLIEST = 1;
}

enum CompressionType {
LZ4 = 0;
NONE = 1;
ZLIB = 2;
ZSTD = 3;
SNAPPY = 4;
}

message Resources {
double cpu = 1;
int64 ram = 2;
Expand Down Expand Up @@ -112,6 +120,7 @@ message ProducerSpec {
bool useThreadLocalProducers = 3;
CryptoSpec cryptoSpec = 4;
string batchBuilder = 5;
CompressionType compressionType = 6;
}

message CryptoSpec {
Expand Down
39 changes: 39 additions & 0 deletions pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
import lombok.extern.slf4j.Slf4j;
import net.jodah.typetools.TypeResolver;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.impl.MessageIdImpl;
Expand Down Expand Up @@ -568,4 +569,42 @@ public static SubscriptionInitialPosition convertFromFunctionDetailsSubscription
return SubscriptionInitialPosition.Latest;
}
}

public static CompressionType convertFromFunctionDetailsCompressionType(
org.apache.pulsar.functions.proto.Function.CompressionType compressionType) {
if (compressionType == null) {
return CompressionType.LZ4;
}
switch (compressionType) {
case NONE:
return CompressionType.NONE;
case ZLIB:
return CompressionType.ZLIB;
case ZSTD:
return CompressionType.ZSTD;
case SNAPPY:
return CompressionType.SNAPPY;
default:
return CompressionType.LZ4;
}
}

public static org.apache.pulsar.functions.proto.Function.CompressionType convertFromCompressionType(
CompressionType compressionType) {
if (compressionType == null) {
return org.apache.pulsar.functions.proto.Function.CompressionType.LZ4;
}
switch (compressionType) {
case NONE:
return org.apache.pulsar.functions.proto.Function.CompressionType.NONE;
case ZLIB:
return org.apache.pulsar.functions.proto.Function.CompressionType.ZLIB;
case ZSTD:
return org.apache.pulsar.functions.proto.Function.CompressionType.ZSTD;
case SNAPPY:
return org.apache.pulsar.functions.proto.Function.CompressionType.SNAPPY;
default:
return org.apache.pulsar.functions.proto.Function.CompressionType.LZ4;
}
}
}
8 changes: 8 additions & 0 deletions ...-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@
import static org.apache.commons.lang3.StringUtils.isEmpty;
import static org.apache.pulsar.common.functions.Utils.BUILTIN;
import static org.apache.pulsar.common.util.ClassLoaderUtils.loadJar;
import static org.apache.pulsar.functions.utils.FunctionCommon.convertFromCompressionType;
import static org.apache.pulsar.functions.utils.FunctionCommon.convertFromFunctionDetailsCompressionType;
import static org.apache.pulsar.functions.utils.FunctionCommon.convertFromFunctionDetailsSubscriptionPosition;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
Expand Down Expand Up @@ -271,6 +273,11 @@ public static FunctionDetails convert(FunctionConfig functionConfig, ExtractedFu
if (producerConf.getBatchBuilder() != null) {
pbldr.setBatchBuilder(producerConf.getBatchBuilder());
}
if (producerConf.getCompressionType() != null) {
pbldr.setCompressionType(convertFromCompressionType(producerConf.getCompressionType()));
} else {
pbldr.setCompressionType(Function.CompressionType.LZ4);
}
sinkSpecBuilder.setProducerSpec(pbldr.build());
}
functionDetailsBuilder.setSink(sinkSpecBuilder);
Expand Down Expand Up @@ -471,6 +478,7 @@ public static FunctionConfig convertFromDetails(FunctionDetails functionDetails)
producerConfig.setBatchBuilder(spec.getBatchBuilder());
}
producerConfig.setUseThreadLocalProducers(spec.getUseThreadLocalProducers());
producerConfig.setCompressionType(convertFromFunctionDetailsCompressionType(spec.getCompressionType()));
functionConfig.setProducerConfig(producerConfig);
}
if (!isEmpty(functionDetails.getLogTopic())) {
Expand Down
8 changes: 8 additions & 0 deletions ...ar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@
package org.apache.pulsar.functions.utils;

import static org.apache.commons.lang3.StringUtils.isEmpty;
import static org.apache.pulsar.functions.utils.FunctionCommon.convertFromCompressionType;
import static org.apache.pulsar.functions.utils.FunctionCommon.convertFromFunctionDetailsCompressionType;
import static org.apache.pulsar.functions.utils.FunctionCommon.convertProcessingGuarantee;
import static org.apache.pulsar.functions.utils.FunctionCommon.getSourceType;
import com.fasterxml.jackson.core.type.TypeReference;
Expand Down Expand Up @@ -164,6 +166,11 @@ public static FunctionDetails convert(SourceConfig sourceConfig, ExtractedSource
if (conf.getBatchBuilder() != null) {
pbldr.setBatchBuilder(conf.getBatchBuilder());
}
if (conf.getCompressionType() != null) {
pbldr.setCompressionType(convertFromCompressionType(conf.getCompressionType()));
} else {
pbldr.setCompressionType(Function.CompressionType.LZ4);
}
sinkSpecBuilder.setProducerSpec(pbldr.build());
}

Expand Down Expand Up @@ -264,6 +271,7 @@ public static SourceConfig convertFromDetails(FunctionDetails functionDetails) {
producerConfig.setBatchBuilder(spec.getBatchBuilder());
}
producerConfig.setUseThreadLocalProducers(spec.getUseThreadLocalProducers());
producerConfig.setCompressionType(convertFromFunctionDetailsCompressionType(spec.getCompressionType()));
sourceConfig.setProducerConfig(producerConfig);
}
if (functionDetails.hasResources()) {
Expand Down
3 changes: 3 additions & 0 deletions ...ctions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

import com.google.gson.Gson;

import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.util.JsonFormat;
Expand Down Expand Up @@ -96,6 +97,7 @@ public void testConvertBackFidelity() {
producerConfig.setMaxPendingMessagesAcrossPartitions(1000);
producerConfig.setUseThreadLocalProducers(true);
producerConfig.setBatchBuilder("DEFAULT");
producerConfig.setCompressionType(CompressionType.ZLIB);
functionConfig.setProducerConfig(producerConfig);
Function.FunctionDetails functionDetails = FunctionConfigUtils.convert(functionConfig, (ClassLoader) null);
FunctionConfig convertedConfig = FunctionConfigUtils.convertFromDetails(functionDetails);
Expand Down Expand Up @@ -137,6 +139,7 @@ public void testConvertWindow() {
producerConfig.setMaxPendingMessagesAcrossPartitions(1000);
producerConfig.setUseThreadLocalProducers(true);
producerConfig.setBatchBuilder("KEY_BASED");
producerConfig.setCompressionType(CompressionType.SNAPPY);
functionConfig.setProducerConfig(producerConfig);
Function.FunctionDetails functionDetails = FunctionConfigUtils.convert(functionConfig, (ClassLoader) null);
FunctionConfig convertedConfig = FunctionConfigUtils.convertFromDetails(functionDetails);
Expand Down
2 changes: 2 additions & 0 deletions ...unctions/utils/src/test/java/org/apache/pulsar/functions/utils/SourceConfigUtilsTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.common.functions.FunctionConfig;
import org.apache.pulsar.common.functions.ProducerConfig;
import org.apache.pulsar.common.functions.Resources;
Expand Down Expand Up @@ -370,6 +371,7 @@ private SourceConfig createSourceConfig() {
producerConfig.setMaxPendingMessagesAcrossPartitions(1000);
producerConfig.setUseThreadLocalProducers(true);
producerConfig.setBatchBuilder("DEFAULT");
producerConfig.setCompressionType(CompressionType.ZSTD);
sourceConfig.setProducerConfig(producerConfig);

sourceConfig.setConfigs(configs);
Expand Down

0 comments on commit e0098ee

Please sign in to comment.