You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2022/07/01 11:35:14 UTC
[flink] branch release-1.15 updated (898dc492760 -> 56f8c0ea1f0)
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a change to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
from 898dc492760 [FLINK-28140][python][docs] Improve the documentation by adding Python examples in DataStream API Integration page
new 68bb94630ba [FLINK-26931][Connector/pulsar] Make the producer name and consumer name unique for each instance.
new 56f8c0ea1f0 [FLINK-26931][Connector/Pulsar] Add producer name for testing the unique producer in Pulsar.
The 2 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:
.../pulsar/common/config/PulsarClientFactory.java | 21 ++++++++++++---------
.../pulsar/common/config/PulsarConfiguration.java | 2 +-
.../connector/pulsar/sink/PulsarSinkBuilder.java | 5 +++++
.../pulsar/sink/config/PulsarSinkConfigUtils.java | 6 +++++-
.../connector/pulsar/sink/writer/PulsarWriter.java | 2 +-
.../pulsar/source/PulsarSourceBuilder.java | 7 ++++++-
.../source/config/PulsarSourceConfigUtils.java | 6 +++++-
.../connector/pulsar/sink/PulsarSinkITCase.java | 3 ++-
8 files changed, 37 insertions(+), 15 deletions(-)
[flink] 02/02: [FLINK-26931][Connector/Pulsar] Add producer name for testing the unique producer in Pulsar.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 56f8c0ea1f05d4829188b5213fea2d7567e7ae94
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Fri Apr 29 04:39:35 2022 +0800
[FLINK-26931][Connector/Pulsar] Add producer name for testing the unique producer in Pulsar.
---
.../java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java | 1 +
1 file changed, 1 insertion(+)
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java
index 8d40bb7c237..0be8463a1c9 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java
@@ -75,6 +75,7 @@ class PulsarSinkITCase extends PulsarTestSuiteBase {
.setAdminUrl(operator().adminUrl())
.setDeliveryGuarantee(guarantee)
.setTopics(topic)
+ .setProducerName("pulsar-sink-it-case")
.setSerializationSchema(flinkSchema(new SimpleStringSchema()))
.build();
[flink] 01/02: [FLINK-26931][Connector/pulsar] Make the producer name and consumer name unique for each instance.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 68bb94630ba69ee857f9f88e76108623614638cd
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Wed Mar 30 20:19:51 2022 +0800
[FLINK-26931][Connector/pulsar] Make the producer name and consumer name unique for each instance.
---
.../pulsar/common/config/PulsarClientFactory.java | 21 ++++++++++++---------
.../pulsar/common/config/PulsarConfiguration.java | 2 +-
.../connector/pulsar/sink/PulsarSinkBuilder.java | 5 +++++
.../pulsar/sink/config/PulsarSinkConfigUtils.java | 6 +++++-
.../connector/pulsar/sink/writer/PulsarWriter.java | 2 +-
.../pulsar/source/PulsarSourceBuilder.java | 7 ++++++-
.../source/config/PulsarSourceConfigUtils.java | 6 +++++-
.../connector/pulsar/sink/PulsarSinkITCase.java | 2 +-
8 files changed, 36 insertions(+), 15 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarClientFactory.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarClientFactory.java
index b1214b57c04..a1188126d06 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarClientFactory.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarClientFactory.java
@@ -28,6 +28,7 @@ import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.ProxyProtocol;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import java.util.Map;
import java.util.TreeSet;
@@ -152,8 +153,8 @@ public final class PulsarClientFactory {
}
/**
- * PulsarAdmin shares almost the same configuration with PulsarClient, but we separate this
- * create method for directly creating it.
+ * PulsarAdmin shares almost the same configuration with PulsarClient, we use this
+ * creating method for creating PulsarAdmin instead of using {@link ClientConfigurationData}.
*/
public static PulsarAdmin createAdmin(PulsarConfiguration configuration) {
PulsarAdminBuilder builder = PulsarAdmin.builder();
@@ -200,15 +201,17 @@ public final class PulsarClientFactory {
String authParamsString = configuration.get(PULSAR_AUTH_PARAMS);
return sneakyClient(
() -> AuthenticationFactory.create(authPluginClassName, authParamsString));
- } else if (configuration.contains(PULSAR_AUTH_PARAM_MAP)) {
- Map<String, String> paramsMap = configuration.get(PULSAR_AUTH_PARAM_MAP);
+ } else {
+ Map<String, String> paramsMap = configuration.getProperties(PULSAR_AUTH_PARAM_MAP);
+ if (paramsMap.isEmpty()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "No %s or %s provided",
+ PULSAR_AUTH_PARAMS.key(), PULSAR_AUTH_PARAM_MAP.key()));
+ }
+
return sneakyClient(
() -> AuthenticationFactory.create(authPluginClassName, paramsMap));
- } else {
- throw new IllegalArgumentException(
- String.format(
- "No %s or %s provided",
- PULSAR_AUTH_PARAMS.key(), PULSAR_AUTH_PARAM_MAP.key()));
}
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfiguration.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfiguration.java
index 3e64c6643cd..0681a3ef4cb 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfiguration.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfiguration.java
@@ -76,7 +76,7 @@ public abstract class PulsarConfiguration extends UnmodifiableConfiguration {
return properties;
}
- /** Get an option value from the given config, convert it into the a new value instance. */
+ /** Get an option value from the given config, convert it into a new value instance. */
public <F, T> T get(ConfigOption<F> option, Function<F, T> convertor) {
F value = get(option);
if (value != null) {
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilder.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilder.java
index 1668e3d126a..332a27a6028 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilder.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilder.java
@@ -315,6 +315,11 @@ public class PulsarSinkBuilder<IN> {
if (!configBuilder.contains(PULSAR_PRODUCER_NAME)) {
LOG.warn(
"We recommend set a readable producer name through setProducerName(String) in production mode.");
+ } else {
+ String producerName = configBuilder.get(PULSAR_PRODUCER_NAME);
+ if (!producerName.contains("%s")) {
+ configBuilder.override(PULSAR_PRODUCER_NAME, producerName + " - %s");
+ }
}
checkNotNull(serializationSchema, "serializationSchema must be set.");
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/PulsarSinkConfigUtils.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/PulsarSinkConfigUtils.java
index 13821fe8918..50f0e1521c1 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/PulsarSinkConfigUtils.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/PulsarSinkConfigUtils.java
@@ -27,6 +27,7 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import java.util.Map;
+import java.util.UUID;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -70,7 +71,10 @@ public final class PulsarSinkConfigUtils {
PulsarClient client, Schema<T> schema, SinkConfiguration configuration) {
ProducerBuilder<T> builder = client.newProducer(schema);
- configuration.useOption(PULSAR_PRODUCER_NAME, builder::producerName);
+ configuration.useOption(
+ PULSAR_PRODUCER_NAME,
+ producerName -> String.format(producerName, UUID.randomUUID()),
+ builder::producerName);
configuration.useOption(
PULSAR_SEND_TIMEOUT_MS,
Math::toIntExact,
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
index 1e4113a8d03..dec4f523364 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
@@ -189,7 +189,7 @@ public class PulsarWriter<IN> implements PrecommittingSinkWriter<IN, PulsarCommi
this.pendingMessages -= 1;
}
- @SuppressWarnings("rawtypes")
+ @SuppressWarnings({"rawtypes", "unchecked"})
private TypedMessageBuilder<?> createMessageBuilder(
String topic, Context context, PulsarMessage<?> message) {
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
index 517286d8341..ed68215819f 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
@@ -473,9 +473,14 @@ public final class PulsarSourceBuilder<OUT> {
if (!configBuilder.contains(PULSAR_CONSUMER_NAME)) {
LOG.warn(
"We recommend set a readable consumer name through setConsumerName(String) in production mode.");
+ } else {
+ String consumerName = configBuilder.get(PULSAR_CONSUMER_NAME);
+ if (!consumerName.contains("%s")) {
+ configBuilder.override(PULSAR_CONSUMER_NAME, consumerName + " - %s");
+ }
}
- // Since these implementation could be a lambda, make sure they are serializable.
+ // Since these implementations could be a lambda, make sure they are serializable.
checkState(isSerializable(startCursor), "StartCursor isn't serializable");
checkState(isSerializable(stopCursor), "StopCursor isn't serializable");
checkState(isSerializable(rangeGenerator), "RangeGenerator isn't serializable");
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
index adb8a03df03..602a1577938 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
@@ -29,6 +29,7 @@ import org.apache.pulsar.client.api.Schema;
import java.util.Map;
import java.util.Optional;
+import java.util.UUID;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -106,7 +107,10 @@ public final class PulsarSourceConfigUtils {
configuration.useOption(
PULSAR_MAX_TOTAL_RECEIVER_QUEUE_SIZE_ACROSS_PARTITIONS,
builder::maxTotalReceiverQueueSizeAcrossPartitions);
- configuration.useOption(PULSAR_CONSUMER_NAME, builder::consumerName);
+ configuration.useOption(
+ PULSAR_CONSUMER_NAME,
+ consumerName -> String.format(consumerName, UUID.randomUUID()),
+ builder::consumerName);
configuration.useOption(PULSAR_READ_COMPACTED, builder::readCompacted);
configuration.useOption(PULSAR_PRIORITY_LEVEL, builder::priorityLevel);
configuration.useOption(
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java
index 94e23d7b097..8d40bb7c237 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java
@@ -61,7 +61,7 @@ class PulsarSinkITCase extends PulsarTestSuiteBase {
@ParameterizedTest
@EnumSource(DeliveryGuarantee.class)
void writeRecordsToPulsar(DeliveryGuarantee guarantee) throws Exception {
- // A random topic with partition 1.
+ // A random topic with partition 4.
String topic = randomAlphabetic(8);
operator().createTopic(topic, 4);
int counts = ThreadLocalRandom.current().nextInt(100, 200);