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:52:56 UTC
(pulsar) branch branch-3.3 updated (d43d0cffbca -> ef2f330f97c)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from d43d0cffbca [fix] [client] Fix resource leak in Pulsar Client since HttpLookupService doesn't get closed (#22858)
new 6fbfe91b698 [fix] [broker] broker log a full thread dump when a deadlock is detected in healthcheck every time (#22916)
new 9a6eecace52 [fix][fn] Support compression type and crypto config for all producers in Functions and Connectors (#22950)
new ef2f330f97c [fix][broker] Check the markDeletePosition and calculate the backlog (#22947)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 14 +-
.../pulsar/broker/admin/impl/BrokersBase.java | 2 +-
.../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 ++++
11 files changed, 490 insertions(+), 266 deletions(-)
create mode 100644 pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ProducerBuilderFactory.java
create mode 100644 pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ProducerBuilderFactoryTest.java
(pulsar) 02/03: [fix][fn] Support compression type and crypto config for all producers in Functions and Connectors (#22950)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9a6eecace523a2abee5f5c653b09294a0b36f449
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 6664a00510e..eeeaa8b3627 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;
@@ -154,27 +154,27 @@ class ContextImpl implements Context, SinkContext, SourceContext, AutoCloseable
this.statsManager = statsManager;
this.fatalHandler = fatalHandler;
- 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 {
@@ -556,26 +556,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 6516b9284c9..115ef1e8a3f 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
@@ -103,7 +103,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 ee59317daf7..3c96837e437 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 692d7459268..6229bffff53 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;
@@ -152,29 +151,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) {
@@ -259,23 +236,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 (!isEmpty(functionDetails.getLogTopic())) {
sourceConfig.setLogTopic(functionDetails.getLogTopic());
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);
+ }
}
(pulsar) 03/03: [fix][broker] Check the markDeletePosition and calculate the backlog (#22947)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ef2f330f97cd6b925251e3ccd07d07c201f25c85
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Thu Jun 20 21:47:27 2024 +0800
[fix][broker] Check the markDeletePosition and calculate the backlog (#22947)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit 82b8d98a488191d279612d5cf2b4846627863543)
---
.../apache/bookkeeper/mledger/impl/ManagedCursorImpl.java | 14 +++++++++-----
1 file changed, 9 insertions(+), 5 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 1d2065ef8e3..780f47dfbe2 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -1115,6 +1115,13 @@ public class ManagedCursorImpl implements ManagedCursor {
return ledger.estimateBacklogFromPosition(markDeletePosition);
}
+ private long getNumberOfEntriesInBacklog() {
+ if (markDeletePosition.compareTo(ledger.getLastPosition()) >= 0) {
+ return 0;
+ }
+ return getNumberOfEntries(Range.openClosed(markDeletePosition, ledger.getLastPosition()));
+ }
+
@Override
public long getNumberOfEntriesInBacklog(boolean isPrecise) {
if (log.isDebugEnabled()) {
@@ -1123,16 +1130,13 @@ public class ManagedCursorImpl implements ManagedCursor {
messagesConsumedCounter, markDeletePosition, readPosition);
}
if (isPrecise) {
- if (markDeletePosition.compareTo(ledger.getLastPosition()) >= 0) {
- return 0;
- }
- return getNumberOfEntries(Range.openClosed(markDeletePosition, ledger.getLastPosition()));
+ return getNumberOfEntriesInBacklog();
}
long backlog = ManagedLedgerImpl.ENTRIES_ADDED_COUNTER_UPDATER.get(ledger) - messagesConsumedCounter;
if (backlog < 0) {
// In some case the counters get incorrect values, fall back to the precise backlog count
- backlog = getNumberOfEntries(Range.openClosed(markDeletePosition, ledger.getLastPosition()));
+ backlog = getNumberOfEntriesInBacklog();
}
return backlog;
(pulsar) 01/03: [fix] [broker] broker log a full thread dump when a deadlock is detected in healthcheck every time (#22916)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6fbfe91b698ccb3f7768e62c7c61d50ab76f2cc0
Author: yangyijun <10...@qq.com>
AuthorDate: Thu Jun 20 18:43:43 2024 +0800
[fix] [broker] broker log a full thread dump when a deadlock is detected in healthcheck every time (#22916)
(cherry picked from commit ca6450598469f158d8fa4cc942fb51e12ed1b609)
---
.../src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
index 7eeea66db71..10b15f6175f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
@@ -88,7 +88,7 @@ public class BrokersBase extends AdminResource {
private static final Duration HEALTH_CHECK_READ_TIMEOUT = Duration.ofSeconds(58);
private static final TimeoutException HEALTH_CHECK_TIMEOUT_EXCEPTION =
FutureUtil.createTimeoutException("Timeout", BrokersBase.class, "healthCheckRecursiveReadNext(...)");
- private volatile long threadDumpLoggedTimestamp;
+ private static volatile long threadDumpLoggedTimestamp;
@GET
@Path("/{cluster}")