This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new d1701ee6e0 [Feature][Connector-V2] Support Pulsar sink multi-table
writes (#10670)
d1701ee6e0 is described below
commit d1701ee6e085c7cd1de00dda7efe29ac481d1511
Author: M MUKTHANANDA REDDDY <[email protected]>
AuthorDate: Fri Jun 19 19:21:19 2026 +0530
[Feature][Connector-V2] Support Pulsar sink multi-table writes (#10670)
---
docs/en/connectors/sink/Pulsar.md | 25 ++-
.../seatunnel/pulsar/sink/PulsarSink.java | 13 +-
.../seatunnel/pulsar/sink/PulsarSinkFactory.java | 20 +-
.../seatunnel/pulsar/sink/PulsarSinkWriter.java | 240 +++++++++++++++++----
.../pulsar/sink/PulsarSinkFactoryTest.java | 91 ++++++++
.../pulsar/sink/PulsarSinkWriterTest.java | 240 +++++++++++++++++++++
6 files changed, 578 insertions(+), 51 deletions(-)
diff --git a/docs/en/connectors/sink/Pulsar.md
b/docs/en/connectors/sink/Pulsar.md
index e67086c693..924007edfe 100644
--- a/docs/en/connectors/sink/Pulsar.md
+++ b/docs/en/connectors/sink/Pulsar.md
@@ -28,7 +28,7 @@ Sink connector for Apache Pulsar.
| Name | Type | Required | Default |
Description
|
|----------------------|--------|----------|---------------------|------------------------------------------------------------------------------------------------------------------|
-| topic | String | Yes | - | sink pulsar
topic
|
+| topic | String | No | - | Sink Pulsar
topic. Required for single-table writes and optional when records provide
`SeaTunnelRow.tableId`. |
| client.service-url | String | Yes | - | Service URL
provider for Pulsar service.
|
| admin.service-url | String | Yes | - | The Pulsar
service HTTP URL for the admin endpoint.
|
| auth.plugin-class | String | No | - | Name of the
authentication plugin.
|
@@ -77,6 +77,15 @@ If you customize the delimiter, add the "field_delimiter"
option.
Customize the field delimiter for data format.The default field_delimiter is
','.
+### topic [String]
+
+The default Pulsar topic used by the sink.
+
+For single-table pipelines, this option is required.
+For multi-table pipelines, the sink will use `SeaTunnelRow.getTableId()` as
the target topic when it is present, and fall back to `topic` only when the row
does not carry a table id.
+
+If neither `SeaTunnelRow.getTableId()` nor `topic` is available, the sink
fails fast with a configuration error.
+
### semantics [Enum]
Consistency semantics for writing to pulsar.
@@ -171,6 +180,20 @@ sink {
}
```
+### Multi-table
+
+> This example routes each row to the Pulsar topic carried in
`SeaTunnelRow.tableId`. In this mode, `topic` can be omitted.
+
+```hocon
+sink {
+ Pulsar {
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ plugin_output = "test"
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSink.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSink.java
index ef5aeb0c57..0826feefca 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSink.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSink.java
@@ -23,6 +23,7 @@ import org.apache.seatunnel.api.serialization.Serializer;
import org.apache.seatunnel.api.sink.SeaTunnelSink;
import org.apache.seatunnel.api.sink.SinkCommitter;
import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.sink.SupportMultiTableSink;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
@@ -42,7 +43,11 @@ import java.util.Optional;
*/
public class PulsarSink
implements SeaTunnelSink<
- SeaTunnelRow, PulsarSinkState, PulsarCommitInfo,
PulsarAggregatedCommitInfo> {
+ SeaTunnelRow,
+ PulsarSinkState,
+ PulsarCommitInfo,
+ PulsarAggregatedCommitInfo>,
+ SupportMultiTableSink {
private final SeaTunnelRowType seaTunnelRowType;
private final PulsarClientConfig clientConfig;
@@ -68,7 +73,11 @@ public class PulsarSink
public SinkWriter<SeaTunnelRow, PulsarCommitInfo, PulsarSinkState>
createWriter(
SinkWriter.Context context) {
return new PulsarSinkWriter(
- context, clientConfig, seaTunnelRowType, readonlyConfig,
Collections.emptyList());
+ context,
+ clientConfig,
+ seaTunnelRowType,
+ readonlyConfig,
+ Collections.<PulsarSinkState>emptyList());
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactory.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactory.java
index 6317c55f76..5db5575c69 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactory.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactory.java
@@ -17,12 +17,15 @@
package org.apache.seatunnel.connectors.seatunnel.pulsar.sink;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.table.connector.TableSink;
import org.apache.seatunnel.api.table.factory.Factory;
import org.apache.seatunnel.api.table.factory.TableSinkFactory;
import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
+import org.apache.seatunnel.common.exception.CommonErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarSinkOptions;
+import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException;
import com.google.auto.service.AutoService;
@@ -36,11 +39,9 @@ public class PulsarSinkFactory implements TableSinkFactory {
@Override
public OptionRule optionRule() {
return OptionRule.builder()
- .required(
- PulsarSinkOptions.CLIENT_SERVICE_URL,
- PulsarSinkOptions.ADMIN_SERVICE_URL,
- PulsarSinkOptions.TOPIC)
+ .required(PulsarSinkOptions.CLIENT_SERVICE_URL,
PulsarSinkOptions.ADMIN_SERVICE_URL)
.optional(
+ PulsarSinkOptions.TOPIC,
PulsarSinkOptions.FORMAT,
PulsarSinkOptions.FIELD_DELIMITER,
PulsarSinkOptions.MESSAGE_ROUTING_MODE,
@@ -58,6 +59,17 @@ public class PulsarSinkFactory implements TableSinkFactory {
@Override
public TableSink createSink(TableSinkFactoryContext context) {
+ validateSingleTableTopic(context);
return () -> new PulsarSink(context.getOptions(),
context.getCatalogTable());
}
+
+ private void validateSingleTableTopic(TableSinkFactoryContext context) {
+ ReadonlyConfig options = context.getOptions();
+ if (context.getCatalogTable() != null
+ && !options.getOptional(PulsarSinkOptions.TOPIC).isPresent()) {
+ throw new PulsarConnectorException(
+ CommonErrorCode.ILLEGAL_ARGUMENT,
+ "Topic must be configured for single-table Pulsar sink.");
+ }
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
index 26f35fab99..22e3c69448 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
@@ -21,7 +21,9 @@ import
org.apache.seatunnel.shade.com.google.common.collect.Lists;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.serialization.SerializationSchema;
+import org.apache.seatunnel.api.sink.MultiTableResourceManager;
import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.sink.SupportMultiTableSinkWriter;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
@@ -47,26 +49,42 @@ import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.TypedMessageBuilder;
import org.apache.pulsar.client.impl.transaction.TransactionImpl;
+import lombok.extern.slf4j.Slf4j;
+
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
+@Slf4j
public class PulsarSinkWriter
- implements SinkWriter<SeaTunnelRow, PulsarCommitInfo, PulsarSinkState>
{
+ implements SinkWriter<SeaTunnelRow, PulsarCommitInfo, PulsarSinkState>,
+ SupportMultiTableSinkWriter<Void> {
+
+ @FunctionalInterface
+ interface ProducerCreator {
+ Producer<byte[]> create(String topic) throws PulsarClientException;
+ }
- private Producer<byte[]> producer;
+ private final Map<String, Producer<byte[]>> producerMap = new
ConcurrentHashMap<>();
+ private final AtomicLong pendingMessages;
+ private final AtomicReference<Throwable> sendMessageException;
+ private final ReadonlyConfig pluginConfig;
+ private final MessageRoutingMode messageRoutingMode;
+ private final ProducerCreator producerCreator;
private PulsarClient pulsarClient;
private SerializationSchema serializationSchema;
private SerializationSchema keySerializationSchema;
- private TransactionImpl transaction;
+ private volatile TransactionImpl transaction;
private int transactionTimeout;
private PulsarSemantics pulsarSemantics;
- private final AtomicLong pendingMessages;
public PulsarSinkWriter(
Context context,
@@ -74,51 +92,104 @@ public class PulsarSinkWriter
SeaTunnelRowType seaTunnelRowType,
ReadonlyConfig pluginConfig,
List<PulsarSinkState> pulsarStates) {
- String topic = pluginConfig.get(PulsarSinkOptions.TOPIC);
+ this(
+ seaTunnelRowType,
+ pluginConfig,
+ pulsarStates,
+ PulsarConfigUtil.createClient(
+ clientConfig,
pluginConfig.get(PulsarSinkOptions.SEMANTICS)),
+ null);
+ }
+
+ PulsarSinkWriter(
+ SeaTunnelRowType seaTunnelRowType,
+ ReadonlyConfig pluginConfig,
+ List<PulsarSinkState> pulsarStates,
+ PulsarClient pulsarClient,
+ ProducerCreator producerCreator) {
String format = pluginConfig.get(PulsarSinkOptions.FORMAT);
String delimiter = pluginConfig.get(PulsarSinkOptions.FIELD_DELIMITER);
this.transactionTimeout =
pluginConfig.get(PulsarSinkOptions.TRANSACTION_TIMEOUT);
this.pulsarSemantics = pluginConfig.get(PulsarSinkOptions.SEMANTICS);
- MessageRoutingMode messageRoutingMode =
- pluginConfig.get(PulsarSinkOptions.MESSAGE_ROUTING_MODE);
+ this.messageRoutingMode =
pluginConfig.get(PulsarSinkOptions.MESSAGE_ROUTING_MODE);
this.serializationSchema = createSerializationSchema(seaTunnelRowType,
format, delimiter);
List<String> partitionKeyList = getPartitionKeyFields(pluginConfig,
seaTunnelRowType);
this.keySerializationSchema =
createKeySerializationSchema(partitionKeyList,
seaTunnelRowType);
- this.pulsarClient = PulsarConfigUtil.createClient(clientConfig,
pulsarSemantics);
+ this.pulsarClient = pulsarClient;
+ this.pluginConfig = pluginConfig;
+ this.producerCreator =
+ producerCreator != null
+ ? producerCreator
+ : topic ->
+ PulsarConfigUtil.createProducer(
+ this.pulsarClient,
+ topic,
+ pulsarSemantics,
+ pluginConfig,
+ messageRoutingMode);
+ this.pendingMessages = new AtomicLong(0);
+ this.sendMessageException = new AtomicReference<>();
if (PulsarSemantics.EXACTLY_ONCE == pulsarSemantics) {
- try {
- this.transaction =
- (TransactionImpl)
- PulsarConfigUtil.getTransaction(pulsarClient,
transactionTimeout);
- } catch (Exception e) {
- throw new PulsarConnectorException(
- PulsarConnectorErrorCode.CREATE_TRANSACTION_FAILED,
- "Pulsar transaction create fail.");
- }
+ this.transaction = createTransaction();
+ }
+ }
+
+ String resolveTopic(SeaTunnelRow row) {
+ String tableId = row.getTableId();
+ if (tableId != null && !tableId.isEmpty()) {
+ return tableId;
}
+
+ String topic = pluginConfig.get(PulsarSinkOptions.TOPIC);
+ if (topic == null || topic.isEmpty()) {
+ throw new PulsarConnectorException(
+ CommonErrorCode.ILLEGAL_ARGUMENT,
+ "Topic must be configured when SeaTunnelRow.getTableId()
is null");
+ }
+
+ return topic;
+ }
+
+ Producer<byte[]> getOrCreateProducer(String topic) {
+ Producer<byte[]> existing = producerMap.get(topic);
+ if (existing != null) {
+ return existing;
+ }
+
try {
- this.producer =
- PulsarConfigUtil.createProducer(
- pulsarClient, topic, pulsarSemantics,
pluginConfig, messageRoutingMode);
+ Producer<byte[]> producer = producerCreator.create(topic);
+
+ producerMap.put(topic, producer);
+ return producer;
+
} catch (PulsarClientException e) {
throw new PulsarConnectorException(
PulsarConnectorErrorCode.CREATE_PRODUCER_FAILED,
- "Pulsar Producer create fail.");
+ "Failed to create Pulsar producer for topic: " + topic,
+ e);
}
- this.pendingMessages = new AtomicLong(0);
}
@Override
public void write(SeaTunnelRow element) throws IOException {
+ checkSendException();
+
+ String topic = resolveTopic(element);
byte[] message = serializationSchema.serialize(element);
byte[] key = null;
if (keySerializationSchema != null) {
key = keySerializationSchema.serialize(element);
}
+
+ Producer<byte[]> topicProducer = getOrCreateProducer(topic);
+
TypedMessageBuilder<byte[]> typedMessageBuilder =
- PulsarConfigUtil.createTypedMessageBuilder(producer,
transaction);
+ PulsarConfigUtil.createTypedMessageBuilder(
+ topicProducer,
+ PulsarSemantics.EXACTLY_ONCE == pulsarSemantics ?
transaction : null);
+
if (key != null) {
typedMessageBuilder.keyBytes(key);
}
@@ -132,9 +203,8 @@ public class PulsarSinkWriter
(id, ex) -> {
pendingMessages.decrementAndGet();
if (ex != null) {
- throw new PulsarConnectorException(
-
PulsarConnectorErrorCode.SEND_MESSAGE_FAILED,
- "send message failed");
+ log.error("Failed to send message to topic {}",
topic, ex);
+ sendMessageException.compareAndSet(null, ex);
}
});
}
@@ -142,7 +212,8 @@ public class PulsarSinkWriter
@Override
public Optional<PulsarCommitInfo> prepareCommit() throws IOException {
- if (PulsarSemantics.EXACTLY_ONCE == pulsarSemantics) {
+ checkSendException();
+ if (PulsarSemantics.EXACTLY_ONCE == pulsarSemantics && transaction !=
null) {
PulsarCommitInfo pulsarCommitInfo = new
PulsarCommitInfo(this.transaction.getTxnID());
return Optional.of(pulsarCommitInfo);
} else {
@@ -153,40 +224,52 @@ public class PulsarSinkWriter
@Override
public List<PulsarSinkState> snapshotState(long checkpointId) throws
IOException {
if (PulsarSemantics.NON != pulsarSemantics) {
- /** flush pending messages */
- producer.flush();
- while (pendingMessages.longValue() > 0) {
- producer.flush();
- }
+ flushPendingMessages();
+ checkSendException();
}
if (PulsarSemantics.EXACTLY_ONCE == pulsarSemantics) {
List<PulsarSinkState> pulsarSinkStates =
Lists.newArrayList(new
PulsarSinkState(this.transaction.getTxnID()));
- try {
- this.transaction =
- (TransactionImpl)
- PulsarConfigUtil.getTransaction(pulsarClient,
transactionTimeout);
- } catch (Exception e) {
- throw new PulsarConnectorException(
- PulsarConnectorErrorCode.CREATE_TRANSACTION_FAILED,
- "Pulsar transaction create fail.");
- }
+ this.transaction = createTransaction();
return pulsarSinkStates;
}
+
return Collections.emptyList();
}
@Override
public void abortPrepare() {
- if (PulsarSemantics.EXACTLY_ONCE == pulsarSemantics) {
+ if (PulsarSemantics.EXACTLY_ONCE == pulsarSemantics && transaction !=
null) {
transaction.abort();
}
}
@Override
public void close() throws IOException {
- producer.close();
- pulsarClient.close();
+ Throwable closeFailure = null;
+ for (Producer<byte[]> producer : producerMap.values()) {
+ try {
+ producer.close();
+ } catch (Throwable throwable) {
+ closeFailure = appendSuppressed(closeFailure, throwable);
+ }
+ }
+ if (pulsarClient != null) {
+ try {
+ pulsarClient.close();
+ } catch (Throwable throwable) {
+ closeFailure = appendSuppressed(closeFailure, throwable);
+ }
+ }
+
+ Throwable sendFailure = sendMessageException.get();
+ if (sendFailure != null) {
+ closeFailure = appendSuppressed(closeFailure,
buildSendFailureException(sendFailure));
+ }
+
+ if (closeFailure != null) {
+ rethrowCloseFailure(closeFailure);
+ }
}
private SerializationSchema createSerializationSchema(
@@ -251,4 +334,73 @@ public class PulsarSinkWriter
}
return Collections.emptyList();
}
+
+ private TransactionImpl createTransaction() {
+ try {
+ return (TransactionImpl)
+ PulsarConfigUtil.getTransaction(pulsarClient,
transactionTimeout);
+ } catch (Exception e) {
+ throw new PulsarConnectorException(
+ PulsarConnectorErrorCode.CREATE_TRANSACTION_FAILED,
+ "Pulsar transaction create fail.",
+ e);
+ }
+ }
+
+ private void flushPendingMessages() throws IOException {
+ for (Producer<byte[]> producer : producerMap.values()) {
+ producer.flush();
+ }
+
+ while (pendingMessages.longValue() > 0) {
+ checkSendException();
+ for (Producer<byte[]> producer : producerMap.values()) {
+ producer.flush();
+ }
+ }
+ }
+
+ private void checkSendException() {
+ Throwable throwable = sendMessageException.get();
+ if (throwable != null) {
+ throw buildSendFailureException(throwable);
+ }
+ }
+
+ private PulsarConnectorException buildSendFailureException(Throwable
throwable) {
+ return new PulsarConnectorException(
+ PulsarConnectorErrorCode.SEND_MESSAGE_FAILED,
+ "Send message failed, please check previous error log for
details.",
+ throwable);
+ }
+
+ private Throwable appendSuppressed(Throwable existingFailure, Throwable
newFailure) {
+ if (existingFailure == null) {
+ return newFailure;
+ }
+ existingFailure.addSuppressed(newFailure);
+ return existingFailure;
+ }
+
+ private void rethrowCloseFailure(Throwable throwable) throws IOException {
+ if (throwable instanceof IOException) {
+ throw (IOException) throwable;
+ }
+ if (throwable instanceof RuntimeException) {
+ throw (RuntimeException) throwable;
+ }
+ throw new IOException("Failed to close Pulsar sink writer.",
throwable);
+ }
+
+ @Override
+ public MultiTableResourceManager<Void> initMultiTableResourceManager(
+ int tableSize, int queueSize) {
+ return null;
+ }
+
+ @Override
+ public void setMultiTableResourceManager(
+ MultiTableResourceManager<Void> multiTableResourceManager, int
queueIndex) {
+ // Pulsar sink does not require shared resources across tables
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactoryTest.java
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactoryTest.java
new file mode 100644
index 0000000000..69ba2888f2
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactoryTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.pulsar.sink;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
+import org.apache.seatunnel.api.table.type.BasicType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class PulsarSinkFactoryTest {
+
+ @Test
+ public void testCreateSinkRequiresTopicForSingleTable() {
+ PulsarSinkFactory factory = new PulsarSinkFactory();
+
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ factory.createSink(
+ new TableSinkFactoryContext(
+ getCatalogTable(), config(),
getClass().getClassLoader())));
+ }
+
+ @Test
+ public void testCreateSinkAllowsMissingTopicForMultiTable() {
+ PulsarSinkFactory factory = new PulsarSinkFactory();
+
+ assertDoesNotThrow(
+ () ->
+ factory.createSink(
+ new TableSinkFactoryContext(
+ null, config(),
getClass().getClassLoader())));
+ }
+
+ private ReadonlyConfig config() {
+ Map<String, Object> options = new HashMap<>();
+ options.put("client.service-url", "pulsar://localhost:6650");
+ options.put("admin.service-url", "http://localhost:8080";);
+ options.put("format", "json");
+ options.put("field_delimiter", ",");
+ options.put("semantics", "NON");
+ options.put("message.routing.mode", "RoundRobinPartition");
+ return ReadonlyConfig.fromMap(options);
+ }
+
+ private CatalogTable getCatalogTable() {
+ List<Column> columns = new ArrayList<>();
+ columns.add(
+ PhysicalColumn.builder()
+ .name("id")
+ .dataType(BasicType.INT_TYPE)
+ .nullable(true)
+ .build());
+ TableSchema tableSchema =
TableSchema.builder().columns(columns).build();
+ return CatalogTable.of(
+ TableIdentifier.of("default", "default", "pulsar_table"),
+ tableSchema,
+ new HashMap<>(),
+ new ArrayList<>(),
+ "Pulsar table");
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriterTest.java
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriterTest.java
new file mode 100644
index 0000000000..e26dd9b6d0
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriterTest.java
@@ -0,0 +1,240 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.pulsar.sink;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException;
+
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Proxy;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class PulsarSinkWriterTest {
+
+ private PulsarSinkWriter createWriter(ReadonlyConfig config) {
+ return new PulsarSinkWriter(
+ new SeaTunnelRowType(new String[] {}, new SeaTunnelDataType[]
{}),
+ config,
+ java.util.Collections.emptyList(),
+ null,
+ topic -> createProducerProxy());
+ }
+
+ @Test
+ public void testResolveTopicWithTableId() {
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {});
+ row.setTableId("persistent://tenant/ns/topic1");
+
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("topic", "fallback-topic");
+ configMap.put("format", "json");
+ configMap.put("field_delimiter", ",");
+ configMap.put("transaction_timeout", 1000);
+ configMap.put("semantics", "NON");
+ configMap.put("message.routing.mode", "RoundRobinPartition");
+
+ ReadonlyConfig config = ReadonlyConfig.fromMap(configMap);
+
+ PulsarSinkWriter writer = createWriter(config);
+
+ String topic = writer.resolveTopic(row);
+
+ assertEquals("persistent://tenant/ns/topic1", topic);
+ }
+
+ @Test
+ public void testResolveTopicWithoutTableId() {
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {});
+ // no tableId set
+
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("topic", "fallback-topic");
+ configMap.put("format", "json");
+ configMap.put("field_delimiter", ",");
+ configMap.put("transaction_timeout", 1000);
+ configMap.put("semantics", "NON");
+ configMap.put("message.routing.mode", "RoundRobinPartition");
+
+ ReadonlyConfig config = ReadonlyConfig.fromMap(configMap);
+
+ PulsarSinkWriter writer = createWriter(config);
+
+ String topic = writer.resolveTopic(row);
+
+ assertEquals("fallback-topic", topic);
+ }
+
+ @Test
+ public void testResolveTopicWithoutTableIdAndWithoutTopic() {
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {});
+
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("format", "json");
+ configMap.put("field_delimiter", ",");
+ configMap.put("transaction_timeout", 1000);
+ configMap.put("semantics", "NON");
+ configMap.put("message.routing.mode", "RoundRobinPartition");
+
+ ReadonlyConfig config = ReadonlyConfig.fromMap(configMap);
+
+ PulsarSinkWriter writer = createWriter(config);
+
+ assertThrows(IllegalArgumentException.class, () ->
writer.resolveTopic(row));
+ }
+
+ @Test
+ public void testGetOrCreateProducerCachesProducerByTopic() {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("format", "json");
+ configMap.put("field_delimiter", ",");
+ configMap.put("transaction_timeout", 1000);
+ configMap.put("semantics", "NON");
+ configMap.put("message.routing.mode", "RoundRobinPartition");
+
+ AtomicInteger createCount = new AtomicInteger();
+ PulsarSinkWriter writer =
+ new PulsarSinkWriter(
+ new SeaTunnelRowType(new String[] {}, new
SeaTunnelDataType[] {}),
+ ReadonlyConfig.fromMap(configMap),
+ java.util.Collections.emptyList(),
+ null,
+ topic -> {
+ createCount.incrementAndGet();
+ return createProducerProxy();
+ });
+
+ Producer<byte[]> first = writer.getOrCreateProducer("topic-a");
+ Producer<byte[]> second = writer.getOrCreateProducer("topic-a");
+ Producer<byte[]> third = writer.getOrCreateProducer("topic-b");
+
+ assertSame(first, second);
+ assertTrue(first != third);
+ assertEquals(2, createCount.get());
+ }
+
+ @Test
+ public void testCloseReleasesResourcesWhenAsyncSendFailed() throws
Exception {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("format", "json");
+ configMap.put("field_delimiter", ",");
+ configMap.put("transaction_timeout", 1000);
+ configMap.put("semantics", "NON");
+ configMap.put("message.routing.mode", "RoundRobinPartition");
+
+ AtomicBoolean producerClosed = new AtomicBoolean(false);
+ AtomicBoolean clientClosed = new AtomicBoolean(false);
+
+ PulsarSinkWriter writer =
+ new PulsarSinkWriter(
+ new SeaTunnelRowType(new String[] {}, new
SeaTunnelDataType[] {}),
+ ReadonlyConfig.fromMap(configMap),
+ java.util.Collections.emptyList(),
+ createPulsarClientProxy(clientClosed),
+ topic -> createProducerProxy(producerClosed));
+ writer.getOrCreateProducer("topic-a");
+
+ Field sendFailureField =
PulsarSinkWriter.class.getDeclaredField("sendMessageException");
+ sendFailureField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ AtomicReference<Throwable> sendFailure =
+ (AtomicReference<Throwable>) sendFailureField.get(writer);
+ sendFailure.set(new RuntimeException("send failed"));
+
+ assertThrows(PulsarConnectorException.class, writer::close);
+ assertTrue(producerClosed.get());
+ assertTrue(clientClosed.get());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Producer<byte[]> createProducerProxy() {
+ return createProducerProxy(new AtomicBoolean(false));
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Producer<byte[]> createProducerProxy(AtomicBoolean closed) {
+ return (Producer<byte[]>)
+ Proxy.newProxyInstance(
+ Producer.class.getClassLoader(),
+ new Class[] {Producer.class},
+ (proxy, method, args) -> {
+ if ("close".equals(method.getName())) {
+ closed.set(true);
+ return null;
+ }
+ Class<?> returnType = method.getReturnType();
+ if (returnType.equals(boolean.class)) {
+ return false;
+ }
+ if (returnType.equals(int.class)) {
+ return 0;
+ }
+ if (returnType.equals(long.class)) {
+ return 0L;
+ }
+ if (returnType.equals(float.class)) {
+ return 0F;
+ }
+ if (returnType.equals(double.class)) {
+ return 0D;
+ }
+ return null;
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static PulsarClient createPulsarClientProxy(AtomicBoolean closed) {
+ return (PulsarClient)
+ Proxy.newProxyInstance(
+ PulsarClient.class.getClassLoader(),
+ new Class[] {PulsarClient.class},
+ (proxy, method, args) -> {
+ if ("close".equals(method.getName())) {
+ closed.set(true);
+ return null;
+ }
+ Class<?> returnType = method.getReturnType();
+ if (returnType.equals(boolean.class)) {
+ return false;
+ }
+ if (returnType.equals(int.class)) {
+ return 0;
+ }
+ if (returnType.equals(long.class)) {
+ return 0L;
+ }
+ return null;
+ });
+ }
+}
