You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/06/20 18:46:46 UTC
(pulsar) 02/03: [fix][fn] Support compression type and crypto config for all producers in Functions and Connectors (#22950)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8935fef2589d73a13007fea001fe993c318a174a
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Jun 20 15:47:25 2024 +0300
[fix][fn] Support compression type and crypto config for all producers in Functions and Connectors (#22950)
(cherry picked from commit ada47a327e08c1866c2b6f102c844e4c83fe93f3)
---
.../pulsar/functions/instance/ContextImpl.java | 65 +++-----
.../functions/instance/ProducerBuilderFactory.java | 159 ++++++++++++++++++
.../apache/pulsar/functions/sink/PulsarSink.java | 158 +++---------------
.../src/main/resources/findbugsExclude.xml | 7 +-
.../pulsar/functions/instance/ContextImplTest.java | 4 +-
.../instance/ProducerBuilderFactoryTest.java | 178 +++++++++++++++++++++
.../functions/utils/FunctionConfigUtils.java | 87 +++++-----
.../pulsar/functions/utils/SourceConfigUtils.java | 47 +-----
.../functions/utils/FunctionConfigUtilsTest.java | 35 ++++
9 files changed, 480 insertions(+), 260 deletions(-)
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
index 5cbbcad24c7..f8b1b3c670f 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
@@ -27,6 +27,7 @@ import io.prometheus.client.Summary;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
@@ -40,14 +41,11 @@ import java.util.stream.Stream;
import lombok.ToString;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.admin.PulsarAdmin;
-import org.apache.pulsar.client.api.BatcherBuilder;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerBuilder;
-import org.apache.pulsar.client.api.HashingScheme;
import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
@@ -55,7 +53,7 @@ import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.api.TypedMessageBuilder;
import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
-import org.apache.pulsar.client.impl.ProducerBuilderImpl;
+import org.apache.pulsar.common.functions.ProducerConfig;
import org.apache.pulsar.common.io.SinkConfig;
import org.apache.pulsar.common.io.SourceConfig;
import org.apache.pulsar.common.naming.TopicName;
@@ -77,6 +75,7 @@ import org.apache.pulsar.functions.secretsprovider.SecretsProvider;
import org.apache.pulsar.functions.source.PulsarFunctionRecord;
import org.apache.pulsar.functions.source.TopicSchema;
import org.apache.pulsar.functions.utils.FunctionCommon;
+import org.apache.pulsar.functions.utils.FunctionConfigUtils;
import org.apache.pulsar.functions.utils.SinkConfigUtils;
import org.apache.pulsar.functions.utils.SourceConfigUtils;
import org.apache.pulsar.io.core.SinkContext;
@@ -88,6 +87,8 @@ import org.slf4j.Logger;
*/
@ToString(exclude = {"pulsarAdmin"})
class ContextImpl implements Context, SinkContext, SourceContext, AutoCloseable {
+ private final ProducerBuilderFactory producerBuilderFactory;
+ private final Map<String, String> producerProperties;
private InstanceConfig config;
private Logger logger;
@@ -99,7 +100,6 @@ class ContextImpl implements Context, SinkContext, SourceContext, AutoCloseable
private final PulsarAdmin pulsarAdmin;
private Map<String, Producer<?>> publishProducers;
private ThreadLocal<Map<String, Producer<?>>> tlPublishProducers;
- private ProducerBuilderImpl<?> producerBuilder;
private final TopicSchema topicSchema;
@@ -151,27 +151,27 @@ class ContextImpl implements Context, SinkContext, SourceContext, AutoCloseable
this.topicSchema = new TopicSchema(client, Thread.currentThread().getContextClassLoader());
this.statsManager = statsManager;
- this.producerBuilder = (ProducerBuilderImpl<?>) client.newProducer().blockIfQueueFull(true).enableBatching(true)
- .batchingMaxPublishDelay(1, TimeUnit.MILLISECONDS);
boolean useThreadLocalProducers = false;
+
Function.ProducerSpec producerSpec = config.getFunctionDetails().getSink().getProducerSpec();
+ ProducerConfig producerConfig = null;
if (producerSpec != null) {
- if (producerSpec.getMaxPendingMessages() != 0) {
- this.producerBuilder.maxPendingMessages(producerSpec.getMaxPendingMessages());
- }
- if (producerSpec.getMaxPendingMessagesAcrossPartitions() != 0) {
- this.producerBuilder
- .maxPendingMessagesAcrossPartitions(producerSpec.getMaxPendingMessagesAcrossPartitions());
- }
- if (producerSpec.getBatchBuilder() != null) {
- if (producerSpec.getBatchBuilder().equals("KEY_BASED")) {
- this.producerBuilder.batcherBuilder(BatcherBuilder.KEY_BASED);
- } else {
- this.producerBuilder.batcherBuilder(BatcherBuilder.DEFAULT);
- }
- }
+ producerConfig = FunctionConfigUtils.convertProducerSpecToProducerConfig(producerSpec);
useThreadLocalProducers = producerSpec.getUseThreadLocalProducers();
}
+ producerBuilderFactory = new ProducerBuilderFactory(client, producerConfig,
+ Thread.currentThread().getContextClassLoader(),
+ // This is for backwards compatibility. The PR https://github.com/apache/pulsar/pull/19470 removed
+ // the default and made it configurable for the producers created in PulsarSink, but not in ContextImpl.
+ // This is to keep the default unchanged for the producers created in ContextImpl.
+ producerBuilder -> producerBuilder.compressionType(CompressionType.LZ4));
+ producerProperties = Collections.unmodifiableMap(InstanceUtils.getProperties(componentType,
+ FunctionCommon.getFullyQualifiedName(
+ this.config.getFunctionDetails().getTenant(),
+ this.config.getFunctionDetails().getNamespace(),
+ this.config.getFunctionDetails().getName()),
+ this.config.getInstanceId()));
+
if (useThreadLocalProducers) {
tlPublishProducers = new ThreadLocal<>();
} else {
@@ -548,26 +548,9 @@ class ContextImpl implements Context, SinkContext, SourceContext, AutoCloseable
}
if (producer == null) {
-
- Producer<T> newProducer = ((ProducerBuilderImpl<T>) producerBuilder.clone())
- .schema(schema)
- .blockIfQueueFull(true)
- .enableBatching(true)
- .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
- .compressionType(CompressionType.LZ4)
- .hashingScheme(HashingScheme.Murmur3_32Hash) //
- .messageRoutingMode(MessageRoutingMode.CustomPartition)
- .messageRouter(FunctionResultRouter.of())
- // set send timeout to be infinity to prevent potential deadlock with consumer
- // that might happen when consumer is blocked due to unacked messages
- .sendTimeout(0, TimeUnit.SECONDS)
- .topic(topicName)
- .properties(InstanceUtils.getProperties(componentType,
- FunctionCommon.getFullyQualifiedName(
- this.config.getFunctionDetails().getTenant(),
- this.config.getFunctionDetails().getNamespace(),
- this.config.getFunctionDetails().getName()),
- this.config.getInstanceId()))
+ Producer<T> newProducer = producerBuilderFactory
+ .createProducerBuilder(topicName, schema, null)
+ .properties(producerProperties)
.create();
if (tlPublishProducers != null) {
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerBuilderFactory.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerBuilderFactory.java
new file mode 100644
index 00000000000..b08f7f3f2cb
--- /dev/null
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerBuilderFactory.java
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.functions.instance;
+
+import static org.apache.commons.lang.StringUtils.isEmpty;
+import com.google.common.annotations.VisibleForTesting;
+import java.security.Security;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import lombok.Builder;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.BatcherBuilder;
+import org.apache.pulsar.client.api.CompressionType;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.HashingScheme;
+import org.apache.pulsar.client.api.MessageRoutingMode;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.functions.CryptoConfig;
+import org.apache.pulsar.common.functions.ProducerConfig;
+import org.apache.pulsar.functions.utils.CryptoUtils;
+import org.bouncycastle.jce.provider.BouncyCastleProvider;
+
+/**
+ * This class is responsible for creating ProducerBuilders with the appropriate configurations to
+ * match the ProducerConfig provided. Producers are created in 2 locations in Pulsar Functions and Connectors
+ * and this class is used to unify the configuration of the producers without duplicating code.
+ */
+@Slf4j
+public class ProducerBuilderFactory {
+
+ private final PulsarClient client;
+ private final ProducerConfig producerConfig;
+ private final Consumer<ProducerBuilder<?>> defaultConfigurer;
+ private final Crypto crypto;
+
+ public ProducerBuilderFactory(PulsarClient client, ProducerConfig producerConfig, ClassLoader functionClassLoader,
+ Consumer<ProducerBuilder<?>> defaultConfigurer) {
+ this.client = client;
+ this.producerConfig = producerConfig;
+ this.defaultConfigurer = defaultConfigurer;
+ try {
+ this.crypto = initializeCrypto(functionClassLoader);
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException("Unable to initialize crypto config " + producerConfig.getCryptoConfig(), e);
+ }
+ if (crypto == null) {
+ log.info("crypto key reader is not provided, not enabling end to end encryption");
+ }
+ }
+
+ public <T> ProducerBuilder<T> createProducerBuilder(String topic, Schema<T> schema, String producerName) {
+ ProducerBuilder<T> builder = client.newProducer(schema);
+ if (defaultConfigurer != null) {
+ defaultConfigurer.accept(builder);
+ }
+ builder.blockIfQueueFull(true)
+ .enableBatching(true)
+ .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
+ .hashingScheme(HashingScheme.Murmur3_32Hash) //
+ .messageRoutingMode(MessageRoutingMode.CustomPartition)
+ .messageRouter(FunctionResultRouter.of())
+ // set send timeout to be infinity to prevent potential deadlock with consumer
+ // that might happen when consumer is blocked due to unacked messages
+ .sendTimeout(0, TimeUnit.SECONDS)
+ .topic(topic);
+ if (producerName != null) {
+ builder.producerName(producerName);
+ }
+ if (producerConfig != null) {
+ if (producerConfig.getCompressionType() != null) {
+ builder.compressionType(producerConfig.getCompressionType());
+ } else {
+ // TODO: address this inconsistency.
+ // PR https://github.com/apache/pulsar/pull/19470 removed the default compression type of LZ4
+ // from the top level. This default is only used if producer config is provided.
+ builder.compressionType(CompressionType.LZ4);
+ }
+ if (producerConfig.getMaxPendingMessages() != null && producerConfig.getMaxPendingMessages() != 0) {
+ builder.maxPendingMessages(producerConfig.getMaxPendingMessages());
+ }
+ if (producerConfig.getMaxPendingMessagesAcrossPartitions() != null
+ && producerConfig.getMaxPendingMessagesAcrossPartitions() != 0) {
+ builder.maxPendingMessagesAcrossPartitions(producerConfig.getMaxPendingMessagesAcrossPartitions());
+ }
+ if (producerConfig.getCryptoConfig() != null) {
+ builder.cryptoKeyReader(crypto.keyReader);
+ builder.cryptoFailureAction(crypto.failureAction);
+ for (String encryptionKeyName : crypto.getEncryptionKeys()) {
+ builder.addEncryptionKey(encryptionKeyName);
+ }
+ }
+ if (producerConfig.getBatchBuilder() != null) {
+ if (producerConfig.getBatchBuilder().equals("KEY_BASED")) {
+ builder.batcherBuilder(BatcherBuilder.KEY_BASED);
+ } else {
+ builder.batcherBuilder(BatcherBuilder.DEFAULT);
+ }
+ }
+ }
+ return builder;
+ }
+
+
+ @SuppressWarnings("unchecked")
+ @VisibleForTesting
+ Crypto initializeCrypto(ClassLoader functionClassLoader) throws ClassNotFoundException {
+ if (producerConfig == null
+ || producerConfig.getCryptoConfig() == null
+ || isEmpty(producerConfig.getCryptoConfig().getCryptoKeyReaderClassName())) {
+ return null;
+ }
+
+ CryptoConfig cryptoConfig = producerConfig.getCryptoConfig();
+
+ // add provider only if it's not in the JVM
+ if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) {
+ Security.addProvider(new BouncyCastleProvider());
+ }
+
+ final String[] encryptionKeys = cryptoConfig.getEncryptionKeys();
+ Crypto.CryptoBuilder bldr = Crypto.builder()
+ .failureAction(cryptoConfig.getProducerCryptoFailureAction())
+ .encryptionKeys(encryptionKeys);
+
+ bldr.keyReader(CryptoUtils.getCryptoKeyReaderInstance(
+ cryptoConfig.getCryptoKeyReaderClassName(), cryptoConfig.getCryptoKeyReaderConfig(),
+ functionClassLoader));
+
+ return bldr.build();
+ }
+
+ @Data
+ @Builder
+ private static class Crypto {
+ private CryptoKeyReader keyReader;
+ private ProducerCryptoFailureAction failureAction;
+ private String[] encryptionKeys;
+ }
+}
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
index 97a0ad0a2ce..18e55e8e84d 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
@@ -18,10 +18,8 @@
*/
package org.apache.pulsar.functions.sink;
-import static org.apache.commons.lang.StringUtils.isEmpty;
import com.google.common.annotations.VisibleForTesting;
import java.nio.charset.StandardCharsets;
-import java.security.Security;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
@@ -29,21 +27,12 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
import java.util.function.Function;
-import lombok.Builder;
-import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
-import org.apache.pulsar.client.api.BatcherBuilder;
-import org.apache.pulsar.client.api.CompressionType;
-import org.apache.pulsar.client.api.CryptoKeyReader;
-import org.apache.pulsar.client.api.HashingScheme;
import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
-import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
@@ -52,22 +41,18 @@ import org.apache.pulsar.client.api.schema.GenericRecord;
import org.apache.pulsar.client.api.schema.KeyValueSchema;
import org.apache.pulsar.client.impl.schema.AutoConsumeSchema;
import org.apache.pulsar.common.functions.ConsumerConfig;
-import org.apache.pulsar.common.functions.CryptoConfig;
import org.apache.pulsar.common.functions.FunctionConfig;
-import org.apache.pulsar.common.functions.ProducerConfig;
import org.apache.pulsar.common.schema.KeyValueEncodingType;
import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.common.util.Reflections;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.functions.instance.AbstractSinkRecord;
-import org.apache.pulsar.functions.instance.FunctionResultRouter;
+import org.apache.pulsar.functions.instance.ProducerBuilderFactory;
import org.apache.pulsar.functions.instance.stats.ComponentStatsManager;
import org.apache.pulsar.functions.source.PulsarRecord;
import org.apache.pulsar.functions.source.TopicSchema;
-import org.apache.pulsar.functions.utils.CryptoUtils;
import org.apache.pulsar.io.core.Sink;
import org.apache.pulsar.io.core.SinkContext;
-import org.bouncycastle.jce.provider.BouncyCastleProvider;
@Slf4j
public class PulsarSink<T> implements Sink<T> {
@@ -82,6 +67,8 @@ public class PulsarSink<T> implements Sink<T> {
PulsarSinkProcessor<T> pulsarSinkProcessor;
private final TopicSchema topicSchema;
+ private Schema<T> schema;
+ private ProducerBuilderFactory producerBuilderFactory;
private interface PulsarSinkProcessor<T> {
@@ -94,60 +81,6 @@ public class PulsarSink<T> implements Sink<T> {
abstract class PulsarSinkProcessorBase implements PulsarSinkProcessor<T> {
protected Map<String, Producer<T>> publishProducers = new ConcurrentHashMap<>();
- protected Schema schema;
- protected Crypto crypto;
-
- protected PulsarSinkProcessorBase(Schema schema, Crypto crypto) {
- this.schema = schema;
- this.crypto = crypto;
- }
-
- public Producer<T> createProducer(PulsarClient client, String topic, String producerName, Schema<T> schema)
- throws PulsarClientException {
- ProducerBuilder<T> builder = client.newProducer(schema)
- .blockIfQueueFull(true)
- .enableBatching(true)
- .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
- .hashingScheme(HashingScheme.Murmur3_32Hash) //
- .messageRoutingMode(MessageRoutingMode.CustomPartition)
- .messageRouter(FunctionResultRouter.of())
- // set send timeout to be infinity to prevent potential deadlock with consumer
- // that might happen when consumer is blocked due to unacked messages
- .sendTimeout(0, TimeUnit.SECONDS)
- .topic(topic);
- if (producerName != null) {
- builder.producerName(producerName);
- }
- 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());
- }
- if (producerConfig.getMaxPendingMessagesAcrossPartitions() != 0) {
- builder.maxPendingMessagesAcrossPartitions(producerConfig.getMaxPendingMessagesAcrossPartitions());
- }
- if (producerConfig.getCryptoConfig() != null) {
- builder.cryptoKeyReader(crypto.keyReader);
- builder.cryptoFailureAction(crypto.failureAction);
- for (String encryptionKeyName : crypto.getEncryptionKeys()) {
- builder.addEncryptionKey(encryptionKeyName);
- }
- }
- if (producerConfig.getBatchBuilder() != null) {
- if (producerConfig.getBatchBuilder().equals("KEY_BASED")) {
- builder.batcherBuilder(BatcherBuilder.KEY_BASED);
- } else {
- builder.batcherBuilder(BatcherBuilder.DEFAULT);
- }
- }
- }
- return builder.properties(properties).create();
- }
protected Producer<T> getProducer(String destinationTopic, Schema schema) {
return getProducer(destinationTopic, null, destinationTopic, schema);
@@ -159,10 +92,9 @@ public class PulsarSink<T> implements Sink<T> {
log.info("Initializing producer {} on topic {} with schema {}",
producerName, topicName, schema);
Producer<T> producer = createProducer(
- client,
topicName,
- producerName,
- schema != null ? schema : this.schema);
+ schema, producerName
+ );
log.info("Initialized producer {} on topic {} with schema {}: {} -> {}",
producerName, topicName, schema, producerId, producer);
return producer;
@@ -218,13 +150,12 @@ public class PulsarSink<T> implements Sink<T> {
@VisibleForTesting
class PulsarSinkAtMostOnceProcessor extends PulsarSinkProcessorBase {
- public PulsarSinkAtMostOnceProcessor(Schema schema, Crypto crypto) {
- super(schema, crypto);
+ public PulsarSinkAtMostOnceProcessor() {
if (!(schema instanceof AutoConsumeSchema)) {
// initialize default topic
try {
publishProducers.put(pulsarSinkConfig.getTopic(),
- createProducer(client, pulsarSinkConfig.getTopic(), null, schema));
+ createProducer(pulsarSinkConfig.getTopic(), schema, null));
} catch (PulsarClientException e) {
log.error("Failed to create Producer while doing user publish", e);
throw new RuntimeException(e);
@@ -270,10 +201,6 @@ public class PulsarSink<T> implements Sink<T> {
@VisibleForTesting
class PulsarSinkAtLeastOnceProcessor extends PulsarSinkAtMostOnceProcessor {
- public PulsarSinkAtLeastOnceProcessor(Schema schema, Crypto crypto) {
- super(schema, crypto);
- }
-
@Override
public void sendOutputMessage(TypedMessageBuilder<T> msg, AbstractSinkRecord<T> record) {
msg.sendAsync()
@@ -284,11 +211,6 @@ public class PulsarSink<T> implements Sink<T> {
@VisibleForTesting
class PulsarSinkManualProcessor extends PulsarSinkAtMostOnceProcessor {
-
- public PulsarSinkManualProcessor(Schema schema, Crypto crypto) {
- super(schema, crypto);
- }
-
@Override
public void sendOutputMessage(TypedMessageBuilder<T> msg, AbstractSinkRecord<T> record) {
super.sendOutputMessage(msg, record);
@@ -297,11 +219,6 @@ public class PulsarSink<T> implements Sink<T> {
@VisibleForTesting
class PulsarSinkEffectivelyOnceProcessor extends PulsarSinkProcessorBase {
-
- public PulsarSinkEffectivelyOnceProcessor(Schema schema, Crypto crypto) {
- super(schema, crypto);
- }
-
@Override
public TypedMessageBuilder<T> newMessage(AbstractSinkRecord<T> record) {
if (!record.getPartitionId().isPresent()) {
@@ -359,30 +276,27 @@ public class PulsarSink<T> implements Sink<T> {
public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
log.info("Opening pulsar sink with config: {}", pulsarSinkConfig);
- Schema<T> schema = initializeSchema();
+ schema = initializeSchema();
if (schema == null) {
log.info("Since output type is null, not creating any real sink");
return;
}
-
- Crypto crypto = initializeCrypto();
- if (crypto == null) {
- log.info("crypto key reader is not provided, not enabling end to end encryption");
- }
+ producerBuilderFactory =
+ new ProducerBuilderFactory(client, pulsarSinkConfig.getProducerConfig(), functionClassLoader, null);
FunctionConfig.ProcessingGuarantees processingGuarantees = this.pulsarSinkConfig.getProcessingGuarantees();
switch (processingGuarantees) {
case ATMOST_ONCE:
- this.pulsarSinkProcessor = new PulsarSinkAtMostOnceProcessor(schema, crypto);
+ this.pulsarSinkProcessor = new PulsarSinkAtMostOnceProcessor();
break;
case ATLEAST_ONCE:
- this.pulsarSinkProcessor = new PulsarSinkAtLeastOnceProcessor(schema, crypto);
+ this.pulsarSinkProcessor = new PulsarSinkAtLeastOnceProcessor();
break;
case EFFECTIVELY_ONCE:
- this.pulsarSinkProcessor = new PulsarSinkEffectivelyOnceProcessor(schema, crypto);
+ this.pulsarSinkProcessor = new PulsarSinkEffectivelyOnceProcessor();
break;
case MANUAL:
- this.pulsarSinkProcessor = new PulsarSinkManualProcessor(schema, crypto);
+ this.pulsarSinkProcessor = new PulsarSinkManualProcessor();
break;
}
}
@@ -427,6 +341,16 @@ public class PulsarSink<T> implements Sink<T> {
}
}
+ Producer<T> createProducer(String topic, Schema<T> schema, String producerName)
+ throws PulsarClientException {
+ ProducerBuilder<T> builder =
+ producerBuilderFactory.createProducerBuilder(topic, schema != null ? schema : this.schema,
+ producerName);
+ return builder
+ .properties(properties)
+ .create();
+ }
+
@SuppressWarnings("unchecked")
@VisibleForTesting
Schema<T> initializeSchema() throws ClassNotFoundException {
@@ -461,39 +385,5 @@ public class PulsarSink<T> implements Sink<T> {
}
}
- @SuppressWarnings("unchecked")
- @VisibleForTesting
- Crypto initializeCrypto() throws ClassNotFoundException {
- if (pulsarSinkConfig.getProducerConfig() == null
- || pulsarSinkConfig.getProducerConfig().getCryptoConfig() == null
- || isEmpty(pulsarSinkConfig.getProducerConfig().getCryptoConfig().getCryptoKeyReaderClassName())) {
- return null;
- }
-
- CryptoConfig cryptoConfig = pulsarSinkConfig.getProducerConfig().getCryptoConfig();
-
- // add provider only if it's not in the JVM
- if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) {
- Security.addProvider(new BouncyCastleProvider());
- }
-
- final String[] encryptionKeys = cryptoConfig.getEncryptionKeys();
- Crypto.CryptoBuilder bldr = Crypto.builder()
- .failureAction(cryptoConfig.getProducerCryptoFailureAction())
- .encryptionKeys(encryptionKeys);
- bldr.keyReader(CryptoUtils.getCryptoKeyReaderInstance(
- cryptoConfig.getCryptoKeyReaderClassName(), cryptoConfig.getCryptoKeyReaderConfig(),
- functionClassLoader));
-
- return bldr.build();
- }
-
- @Data
- @Builder
- private static class Crypto {
- private CryptoKeyReader keyReader;
- private ProducerCryptoFailureAction failureAction;
- private String[] encryptionKeys;
- }
}
diff --git a/pulsar-functions/instance/src/main/resources/findbugsExclude.xml b/pulsar-functions/instance/src/main/resources/findbugsExclude.xml
index 7fe247d2ab2..40e3e911123 100644
--- a/pulsar-functions/instance/src/main/resources/findbugsExclude.xml
+++ b/pulsar-functions/instance/src/main/resources/findbugsExclude.xml
@@ -49,7 +49,12 @@
<Bug pattern="MS_PKGPROTECT"/>
</Match>
<Match>
- <Class name="org.apache.pulsar.functions.sink.PulsarSink$Crypto$CryptoBuilder"/>
+ <Class name="org.apache.pulsar.functions.instance.ProducerBuilderFactory$Crypto$CryptoBuilder"/>
+ <Bug pattern="EI_EXPOSE_REP2"/>
+ </Match>
+ <Match>
+ <Class name="org.apache.pulsar.functions.instance.ProducerBuilderFactory"/>
+ <Method name="<init>"/>
<Bug pattern="EI_EXPOSE_REP2"/>
</Match>
<Match>
diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java
index 90f7df37fa1..108d8e4b666 100644
--- a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java
+++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java
@@ -102,7 +102,9 @@ public class ContextImplTest {
client = mock(PulsarClientImpl.class);
ConnectionPool connectionPool = mock(ConnectionPool.class);
when(client.getCnxPool()).thenReturn(connectionPool);
- when(client.newProducer()).thenReturn(new ProducerBuilderImpl(client, Schema.BYTES));
+ when(client.newProducer()).thenAnswer(invocation -> new ProducerBuilderImpl(client, Schema.BYTES));
+ when(client.newProducer(any())).thenAnswer(
+ invocation -> new ProducerBuilderImpl(client, invocation.getArgument(0)));
when(client.createProducerAsync(any(ProducerConfigurationData.class), any(), any()))
.thenReturn(CompletableFuture.completedFuture(producer));
when(client.getSchema(anyString())).thenReturn(CompletableFuture.completedFuture(Optional.empty()));
diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ProducerBuilderFactoryTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ProducerBuilderFactoryTest.java
new file mode 100644
index 00000000000..42940f7e2da
--- /dev/null
+++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ProducerBuilderFactoryTest.java
@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.functions.instance;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.api.BatcherBuilder;
+import org.apache.pulsar.client.api.CompressionType;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.EncryptionKeyInfo;
+import org.apache.pulsar.client.api.HashingScheme;
+import org.apache.pulsar.client.api.MessageRoutingMode;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.functions.CryptoConfig;
+import org.apache.pulsar.common.functions.ProducerConfig;
+import org.mockito.internal.util.MockUtil;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class ProducerBuilderFactoryTest {
+ private PulsarClient pulsarClient;
+ private ProducerBuilder producerBuilder;
+
+ @BeforeMethod
+ public void setup() {
+ pulsarClient = mock(PulsarClient.class);
+
+ producerBuilder = mock(ProducerBuilder.class);
+ doReturn(producerBuilder).when(producerBuilder).blockIfQueueFull(anyBoolean());
+ doReturn(producerBuilder).when(producerBuilder).enableBatching(anyBoolean());
+ doReturn(producerBuilder).when(producerBuilder).batchingMaxPublishDelay(anyLong(), any());
+ doReturn(producerBuilder).when(producerBuilder).compressionType(any());
+ doReturn(producerBuilder).when(producerBuilder).hashingScheme(any());
+ doReturn(producerBuilder).when(producerBuilder).messageRoutingMode(any());
+ doReturn(producerBuilder).when(producerBuilder).messageRouter(any());
+ doReturn(producerBuilder).when(producerBuilder).topic(anyString());
+ doReturn(producerBuilder).when(producerBuilder).producerName(anyString());
+ doReturn(producerBuilder).when(producerBuilder).property(anyString(), anyString());
+ doReturn(producerBuilder).when(producerBuilder).properties(any());
+ doReturn(producerBuilder).when(producerBuilder).sendTimeout(anyInt(), any());
+
+ doReturn(producerBuilder).when(pulsarClient).newProducer();
+ doReturn(producerBuilder).when(pulsarClient).newProducer(any());
+ }
+
+ @AfterMethod
+ public void tearDown() {
+ MockUtil.resetMock(pulsarClient);
+ pulsarClient = null;
+ MockUtil.resetMock(producerBuilder);
+ producerBuilder = null;
+ TestCryptoKeyReader.LAST_INSTANCE = null;
+ }
+
+ @Test
+ public void testCreateProducerBuilder() {
+ ProducerBuilderFactory builderFactory = new ProducerBuilderFactory(pulsarClient, null, null, null);
+ builderFactory.createProducerBuilder("topic", Schema.STRING, "producerName");
+ verifyCommon();
+ verifyNoMoreInteractions(producerBuilder);
+ }
+
+ private void verifyCommon() {
+ verify(pulsarClient).newProducer(Schema.STRING);
+ verify(producerBuilder).blockIfQueueFull(true);
+ verify(producerBuilder).enableBatching(true);
+ verify(producerBuilder).batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS);
+ verify(producerBuilder).hashingScheme(HashingScheme.Murmur3_32Hash);
+ verify(producerBuilder).messageRoutingMode(MessageRoutingMode.CustomPartition);
+ verify(producerBuilder).messageRouter(FunctionResultRouter.of());
+ verify(producerBuilder).sendTimeout(0, TimeUnit.SECONDS);
+ verify(producerBuilder).topic("topic");
+ verify(producerBuilder).producerName("producerName");
+ }
+
+ @Test
+ public void testCreateProducerBuilderWithDefaultConfigurer() {
+ ProducerBuilderFactory builderFactory = new ProducerBuilderFactory(pulsarClient, null, null,
+ builder -> builder.property("key", "value"));
+ builderFactory.createProducerBuilder("topic", Schema.STRING, "producerName");
+ verifyCommon();
+ verify(producerBuilder).property("key", "value");
+ verifyNoMoreInteractions(producerBuilder);
+ }
+
+ @Test
+ public void testCreateProducerBuilderWithSimpleProducerConfig() {
+ ProducerConfig producerConfig = new ProducerConfig();
+ producerConfig.setBatchBuilder("KEY_BASED");
+ ProducerBuilderFactory builderFactory = new ProducerBuilderFactory(pulsarClient, producerConfig, null, null);
+ builderFactory.createProducerBuilder("topic", Schema.STRING, "producerName");
+ verifyCommon();
+ verify(producerBuilder).compressionType(CompressionType.LZ4);
+ verify(producerBuilder).batcherBuilder(BatcherBuilder.KEY_BASED);
+ verifyNoMoreInteractions(producerBuilder);
+ }
+
+ @Test
+ public void testCreateProducerBuilderWithAdvancedProducerConfig() {
+ ProducerConfig producerConfig = new ProducerConfig();
+ producerConfig.setBatchBuilder("KEY_BASED");
+ producerConfig.setCompressionType(CompressionType.SNAPPY);
+ producerConfig.setMaxPendingMessages(5000);
+ producerConfig.setMaxPendingMessagesAcrossPartitions(50000);
+ CryptoConfig cryptoConfig = new CryptoConfig();
+ cryptoConfig.setProducerCryptoFailureAction(ProducerCryptoFailureAction.FAIL);
+ cryptoConfig.setEncryptionKeys(new String[]{"key1", "key2"});
+ cryptoConfig.setCryptoKeyReaderConfig(Map.of("key", "value"));
+ cryptoConfig.setCryptoKeyReaderClassName(TestCryptoKeyReader.class.getName());
+ producerConfig.setCryptoConfig(cryptoConfig);
+ ProducerBuilderFactory builderFactory = new ProducerBuilderFactory(pulsarClient, producerConfig, null, null);
+ builderFactory.createProducerBuilder("topic", Schema.STRING, "producerName");
+ verifyCommon();
+ verify(producerBuilder).compressionType(CompressionType.SNAPPY);
+ verify(producerBuilder).batcherBuilder(BatcherBuilder.KEY_BASED);
+ verify(producerBuilder).maxPendingMessages(5000);
+ verify(producerBuilder).maxPendingMessagesAcrossPartitions(50000);
+ TestCryptoKeyReader lastInstance = TestCryptoKeyReader.LAST_INSTANCE;
+ assertNotNull(lastInstance);
+ assertEquals(lastInstance.configs, cryptoConfig.getCryptoKeyReaderConfig());
+ verify(producerBuilder).cryptoKeyReader(lastInstance);
+ verify(producerBuilder).cryptoFailureAction(ProducerCryptoFailureAction.FAIL);
+ verify(producerBuilder).addEncryptionKey("key1");
+ verify(producerBuilder).addEncryptionKey("key2");
+ verifyNoMoreInteractions(producerBuilder);
+ }
+
+ public static class TestCryptoKeyReader implements CryptoKeyReader {
+ static TestCryptoKeyReader LAST_INSTANCE;
+ Map<String, Object> configs;
+ public TestCryptoKeyReader(Map<String, Object> configs) {
+ this.configs = configs;
+ assert LAST_INSTANCE == null;
+ LAST_INSTANCE = this;
+ }
+
+ @Override
+ public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> metadata) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> metadata) {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
\ No newline at end of file
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java
index a2a1764692e..d7295d133da 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java
@@ -250,29 +250,7 @@ public class FunctionConfigUtils {
sinkSpecBuilder.setTypeClassName(functionConfig.getOutputTypeClassName());
}
if (functionConfig.getProducerConfig() != null) {
- ProducerConfig producerConf = functionConfig.getProducerConfig();
- Function.ProducerSpec.Builder pbldr = Function.ProducerSpec.newBuilder();
- if (producerConf.getMaxPendingMessages() != null) {
- pbldr.setMaxPendingMessages(producerConf.getMaxPendingMessages());
- }
- if (producerConf.getMaxPendingMessagesAcrossPartitions() != null) {
- pbldr.setMaxPendingMessagesAcrossPartitions(producerConf.getMaxPendingMessagesAcrossPartitions());
- }
- if (producerConf.getUseThreadLocalProducers() != null) {
- pbldr.setUseThreadLocalProducers(producerConf.getUseThreadLocalProducers());
- }
- if (producerConf.getCryptoConfig() != null) {
- pbldr.setCryptoSpec(CryptoUtils.convert(producerConf.getCryptoConfig()));
- }
- 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());
+ sinkSpecBuilder.setProducerSpec(convertProducerConfigToProducerSpec(functionConfig.getProducerConfig()));
}
if (functionConfig.getBatchBuilder() != null) {
Function.ProducerSpec.Builder builder = sinkSpecBuilder.getProducerSpec() != null
@@ -463,23 +441,8 @@ public class FunctionConfigUtils {
functionConfig.setOutputSchemaType(functionDetails.getSink().getSchemaType());
}
if (functionDetails.getSink().getProducerSpec() != null) {
- Function.ProducerSpec spec = functionDetails.getSink().getProducerSpec();
- ProducerConfig producerConfig = new ProducerConfig();
- if (spec.getMaxPendingMessages() != 0) {
- producerConfig.setMaxPendingMessages(spec.getMaxPendingMessages());
- }
- if (spec.getMaxPendingMessagesAcrossPartitions() != 0) {
- producerConfig.setMaxPendingMessagesAcrossPartitions(spec.getMaxPendingMessagesAcrossPartitions());
- }
- if (spec.hasCryptoSpec()) {
- producerConfig.setCryptoConfig(CryptoUtils.convertFromSpec(spec.getCryptoSpec()));
- }
- if (spec.getBatchBuilder() != null) {
- producerConfig.setBatchBuilder(spec.getBatchBuilder());
- }
- producerConfig.setUseThreadLocalProducers(spec.getUseThreadLocalProducers());
- producerConfig.setCompressionType(convertFromFunctionDetailsCompressionType(spec.getCompressionType()));
- functionConfig.setProducerConfig(producerConfig);
+ functionConfig.setProducerConfig(
+ convertProducerSpecToProducerConfig(functionDetails.getSink().getProducerSpec()));
}
if (!isEmpty(functionDetails.getLogTopic())) {
functionConfig.setLogTopic(functionDetails.getLogTopic());
@@ -544,6 +507,50 @@ public class FunctionConfigUtils {
return functionConfig;
}
+ public static Function.ProducerSpec convertProducerConfigToProducerSpec(ProducerConfig producerConf) {
+ Function.ProducerSpec.Builder builder = Function.ProducerSpec.newBuilder();
+ if (producerConf.getMaxPendingMessages() != null) {
+ builder.setMaxPendingMessages(producerConf.getMaxPendingMessages());
+ }
+ if (producerConf.getMaxPendingMessagesAcrossPartitions() != null) {
+ builder.setMaxPendingMessagesAcrossPartitions(producerConf.getMaxPendingMessagesAcrossPartitions());
+ }
+ if (producerConf.getUseThreadLocalProducers() != null) {
+ builder.setUseThreadLocalProducers(producerConf.getUseThreadLocalProducers());
+ }
+ if (producerConf.getCryptoConfig() != null) {
+ builder.setCryptoSpec(CryptoUtils.convert(producerConf.getCryptoConfig()));
+ }
+ if (producerConf.getBatchBuilder() != null) {
+ builder.setBatchBuilder(producerConf.getBatchBuilder());
+ }
+ if (producerConf.getCompressionType() != null) {
+ builder.setCompressionType(convertFromCompressionType(producerConf.getCompressionType()));
+ } else {
+ builder.setCompressionType(Function.CompressionType.LZ4);
+ }
+ return builder.build();
+ }
+
+ public static ProducerConfig convertProducerSpecToProducerConfig(Function.ProducerSpec spec) {
+ ProducerConfig producerConfig = new ProducerConfig();
+ if (spec.getMaxPendingMessages() != 0) {
+ producerConfig.setMaxPendingMessages(spec.getMaxPendingMessages());
+ }
+ if (spec.getMaxPendingMessagesAcrossPartitions() != 0) {
+ producerConfig.setMaxPendingMessagesAcrossPartitions(spec.getMaxPendingMessagesAcrossPartitions());
+ }
+ if (spec.hasCryptoSpec()) {
+ producerConfig.setCryptoConfig(CryptoUtils.convertFromSpec(spec.getCryptoSpec()));
+ }
+ if (spec.getBatchBuilder() != null) {
+ producerConfig.setBatchBuilder(spec.getBatchBuilder());
+ }
+ producerConfig.setUseThreadLocalProducers(spec.getUseThreadLocalProducers());
+ producerConfig.setCompressionType(convertFromFunctionDetailsCompressionType(spec.getCompressionType()));
+ return producerConfig;
+ }
+
public static void inferMissingArguments(FunctionConfig functionConfig,
boolean forwardSourceMessagePropertyEnabled) {
if (StringUtils.isEmpty(functionConfig.getName())) {
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
index a6430bbea45..2239f0fcbc2 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
@@ -19,10 +19,10 @@
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 static org.apache.pulsar.functions.utils.FunctionConfigUtils.convertProducerConfigToProducerSpec;
+import static org.apache.pulsar.functions.utils.FunctionConfigUtils.convertProducerSpecToProducerConfig;
import com.fasterxml.jackson.core.type.TypeReference;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
@@ -39,7 +39,6 @@ import net.bytebuddy.description.type.TypeDefinition;
import net.bytebuddy.description.type.TypeDescription;
import net.bytebuddy.pool.TypePool;
import org.apache.commons.lang3.StringUtils;
-import org.apache.pulsar.common.functions.ProducerConfig;
import org.apache.pulsar.common.functions.Resources;
import org.apache.pulsar.common.io.BatchSourceConfig;
import org.apache.pulsar.common.io.ConnectorDefinition;
@@ -149,29 +148,7 @@ public class SourceConfigUtils {
}
if (sourceConfig.getProducerConfig() != null) {
- ProducerConfig conf = sourceConfig.getProducerConfig();
- Function.ProducerSpec.Builder pbldr = Function.ProducerSpec.newBuilder();
- if (conf.getMaxPendingMessages() != null) {
- pbldr.setMaxPendingMessages(conf.getMaxPendingMessages());
- }
- if (conf.getMaxPendingMessagesAcrossPartitions() != null) {
- pbldr.setMaxPendingMessagesAcrossPartitions(conf.getMaxPendingMessagesAcrossPartitions());
- }
- if (conf.getUseThreadLocalProducers() != null) {
- pbldr.setUseThreadLocalProducers(conf.getUseThreadLocalProducers());
- }
- if (conf.getCryptoConfig() != null) {
- pbldr.setCryptoSpec(CryptoUtils.convert(conf.getCryptoConfig()));
- }
- 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());
+ sinkSpecBuilder.setProducerSpec(convertProducerConfigToProducerSpec(sourceConfig.getProducerConfig()));
}
if (sourceConfig.getBatchBuilder() != null) {
@@ -256,23 +233,7 @@ public class SourceConfigUtils {
sourceConfig.setSerdeClassName(sinkSpec.getSerDeClassName());
}
if (sinkSpec.getProducerSpec() != null) {
- Function.ProducerSpec spec = sinkSpec.getProducerSpec();
- ProducerConfig producerConfig = new ProducerConfig();
- if (spec.getMaxPendingMessages() != 0) {
- producerConfig.setMaxPendingMessages(spec.getMaxPendingMessages());
- }
- if (spec.getMaxPendingMessagesAcrossPartitions() != 0) {
- producerConfig.setMaxPendingMessagesAcrossPartitions(spec.getMaxPendingMessagesAcrossPartitions());
- }
- if (spec.hasCryptoSpec()) {
- producerConfig.setCryptoConfig(CryptoUtils.convertFromSpec(spec.getCryptoSpec()));
- }
- if (spec.getBatchBuilder() != null) {
- producerConfig.setBatchBuilder(spec.getBatchBuilder());
- }
- producerConfig.setUseThreadLocalProducers(spec.getUseThreadLocalProducers());
- producerConfig.setCompressionType(convertFromFunctionDetailsCompressionType(spec.getCompressionType()));
- sourceConfig.setProducerConfig(producerConfig);
+ sourceConfig.setProducerConfig(convertProducerSpecToProducerConfig(sinkSpec.getProducerSpec()));
}
if (functionDetails.hasResources()) {
Resources resources = new Resources();
diff --git a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java
index 954eef44a73..cf4e7dd92a8 100644
--- a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java
+++ b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java
@@ -35,9 +35,12 @@ import java.util.HashMap;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.CompressionType;
+import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
+import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.impl.schema.JSONSchema;
import org.apache.pulsar.common.functions.ConsumerConfig;
+import org.apache.pulsar.common.functions.CryptoConfig;
import org.apache.pulsar.common.functions.FunctionConfig;
import org.apache.pulsar.common.functions.ProducerConfig;
import org.apache.pulsar.common.functions.Resources;
@@ -667,4 +670,36 @@ public class FunctionConfigUtilsTest {
convertedConfig = FunctionConfigUtils.convertFromDetails(functionDetails);
assertTrue(convertedConfig.getInputSpecs().get("test-input").isPoolMessages());
}
+
+ @Test
+ public void testConvertProducerSpecToProducerConfigAndBackToProducerSpec() {
+ // given
+ Function.ProducerSpec producerSpec = Function.ProducerSpec.newBuilder()
+ .setBatchBuilder("KEY_BASED")
+ .setCompressionType(Function.CompressionType.ZSTD)
+ .setCryptoSpec(Function.CryptoSpec.newBuilder()
+ .addProducerEncryptionKeyName("key1")
+ .addProducerEncryptionKeyName("key2")
+ .setConsumerCryptoFailureAction(Function.CryptoSpec.FailureAction.DISCARD)
+ .setProducerCryptoFailureAction(Function.CryptoSpec.FailureAction.SEND)
+ .setCryptoKeyReaderClassName("ReaderClassName")
+ .setCryptoKeyReaderConfig("{\"key\":\"value\"}")
+ .build())
+ .build();
+ // when
+ ProducerConfig producerConfig = FunctionConfigUtils.convertProducerSpecToProducerConfig(producerSpec);
+ // then
+ assertEquals(producerConfig.getBatchBuilder(), "KEY_BASED");
+ assertEquals(producerConfig.getCompressionType(), CompressionType.ZSTD);
+ CryptoConfig cryptoConfig = producerConfig.getCryptoConfig();
+ assertEquals(cryptoConfig.getProducerCryptoFailureAction(), ProducerCryptoFailureAction.SEND);
+ assertEquals(cryptoConfig.getConsumerCryptoFailureAction(), ConsumerCryptoFailureAction.DISCARD);
+ assertEquals(cryptoConfig.getEncryptionKeys(), new String[]{"key1", "key2"});
+ assertEquals(cryptoConfig.getCryptoKeyReaderClassName(), "ReaderClassName");
+ // and when
+ // converted back to producer spec
+ Function.ProducerSpec producerSpec2 = FunctionConfigUtils.convertProducerConfigToProducerSpec(producerConfig);
+ // then
+ assertEquals(producerSpec2, producerSpec);
+ }
}