[Connector][Pulsar] Support multi-table sink#10558
Open
Muktha9491 wants to merge 1 commit intoapache:devfrom
Open
[Connector][Pulsar] Support multi-table sink#10558Muktha9491 wants to merge 1 commit intoapache:devfrom
Muktha9491 wants to merge 1 commit intoapache:devfrom
Conversation
Issue 1: Exceptions thrown in whenComplete callback are ineffectiveLocation: future.whenComplete(
(id, ex) -> {
pendingMessages.decrementAndGet();
if (ex != null) {
throw new PulsarConnectorException(
PulsarConnectorErrorCode.SEND_MESSAGE_FAILED,
"Send message failed");
}
});Related Context:
Problem Description:
This is an issue that existed in the original code, but was not fixed in this refactoring. Potential Risks:
Impact Scope:
Severity: MAJOR Improvement Suggestions: future.whenComplete(
(id, ex) -> {
pendingMessages.decrementAndGet();
if (ex != null) {
// Log errors instead of throwing exceptions
log.error("Failed to send message to topic: {}", topic, ex);
// Or use a custom error handling mechanism
errorHandler.handleSendFailure(topic, element, ex);
}
});Issue 2: Removal of Apache License headerLocations:
Problem Description:
This violates Apache project guidelines. Potential Risks:
Impact Scope:
Severity: BLOCKER (Must fix) Improvement Suggestions: Issue 3: Removal of PARTITION_KEY_FIELDS configuration causes breaking changeLocations:
Problem Description:
Potential Risks:
Impact Scope:
Severity: MAJOR Improvement Suggestions:
Issue 4: Race condition in transaction management logic after prepareCommitLocation: @Override
public Optional<PulsarCommitInfo> prepareCommit() throws IOException {
if (PulsarSemantics.EXACTLY_ONCE != pulsarSemantics) {
return Optional.empty();
}
while (pendingMessages.get() > 0) {
Thread.yield();
}
if (currentTransaction == null) {
return Optional.empty();
}
TxnID txnID = currentTransaction.getTxnID();
currentTransaction = null; // ← Set to null
return Optional.of(new PulsarCommitInfo(txnID));
}
private Transaction getOrCreateTransaction() {
if (PulsarSemantics.EXACTLY_ONCE != pulsarSemantics) {
return null;
}
if (currentTransaction == null) { // ← Recreate
try {
currentTransaction =
PulsarConfigUtil.getTransaction(pulsarClient, transactionTimeout);
} catch (Exception e) {
throw new PulsarConnectorException(
PulsarConnectorErrorCode.CREATE_TRANSACTION_FAILED,
"Transaction create failed");
}
}
return currentTransaction;
}Problem Description:
Compared to original implementation: // The original snapshotState creates a new transaction before returning the state
List<PulsarSinkState> pulsarSinkStates = Lists.newArrayList(new PulsarSinkState(this.transaction.getTxnID()));
this.transaction = (TransactionImpl) PulsarConfigUtil.getTransaction(pulsarClient, transactionTimeout);
return pulsarSinkStates;Potential Risks:
Impact Scope:
Severity: CRITICAL Improvement Suggestions: TxnID txnID = currentTransaction.getTxnID();
currentTransaction = null;
// Create the next transaction immediately
try {
currentTransaction = PulsarConfigUtil.getTransaction(pulsarClient, transactionTimeout);
} catch (Exception e) {
throw new PulsarConnectorException(
PulsarConnectorErrorCode.CREATE_TRANSACTION_FAILED,
"Failed to create next transaction after prepareCommit", e);
}
return Optional.of(new PulsarCommitInfo(txnID));Issue 5: TOPIC configuration becomes optional but default value scenarios not sufficiently validatedLocations: private String resolveTopic(SeaTunnelRow row) {
if (row.getTableId() != null) {
return row.getTableId();
}
return pluginConfig.get(PulsarSinkOptions.TOPIC); // ← May return null
}Problem Description:
Potential Risks:
Impact Scope:
Severity: MINOR Improvement Suggestions:
private String resolveTopic(SeaTunnelRow row) {
if (row.getTableId() != null) {
return row.getTableId();
}
String topic = pluginConfig.get(PulsarSinkOptions.TOPIC);
if (topic == null) {
throw new PulsarConnectorException(
PulsarConnectorErrorCode.ILLEGAL_ARGUMENT,
"Topic must be configured when row.getTableId() is null");
}
return topic;
}Issue 6: Using RuntimeException instead of SeaTunnelJsonFormatExceptionLocation: // Original code:
throw new SeaTunnelJsonFormatException(
CommonErrorCode.UNSUPPORTED_DATA_TYPE,
"Unsupported format: " + format);
// New code:
throw new RuntimeException("Unsupported format: " + format);Problem Description: Potential Risks:
Impact Scope:
Severity: MINOR Improvement Suggestions: Issue 7: Lack of testing for multi-table scenariosProblem Description:
Potential Risks:
Impact Scope:
Severity: MAJOR Improvement Suggestions: // PulsarSinkWriterTest.java
@Test
public void testResolveTopicWithTableId() {
SeaTunnelRow row = new SeaTunnelRow(new Object[]{});
row.setTableId("persistent://tenant/ns/topic1");
String topic = writer.resolveTopic(row);
assertEquals("persistent://tenant/ns/topic1", topic);
}
@Test
public void testResolveTopicWithoutTableId() {
SeaTunnelRow row = new SeaTunnelRow(new Object[]{});
String topic = writer.resolveTopic(row);
assertEquals(configTopic, topic);
}
@Test
public void testMultipleProducerCreation() {
// Verify that multiple producers are cached in producerMap
}
@Test
public void testExactlyOnceMultiTable() {
// Verify multi-table writes in EXACTLY_ONCE mode
}Issue 8: SupportMultiTableSinkWriter interface not implementedLocation: public class PulsarSinkWriter
implements SinkWriter<SeaTunnelRow, PulsarCommitInfo, PulsarSinkState> {
// SupportMultiTableSinkWriter not implementedRelated Context:
Problem Description: Potential Risks:
Impact Scope:
Severity: MINOR Improvement Suggestions: Issue 9: snapshotState returning empty list may cause state lossLocation: @Override
public List<PulsarSinkState> snapshotState(long checkpointId) throws IOException {
for (Producer<byte[]> producer : producerMap.values()) {
producer.flush();
}
while (pendingMessages.get() > 0) {
for (Producer<byte[]> producer : producerMap.values()) {
producer.flush();
}
}
return Collections.emptyList(); // ← Return empty list
}Problem Description:
Although SeaTunnel's checkpoint mechanism allows Potential Risks:
Impact Scope:
Severity: MAJOR Improvement Suggestions:
@Override
public List<PulsarSinkState> snapshotState(long checkpointId) throws IOException {
// flush logic...
if (PulsarSemantics.EXACTLY_ONCE == pulsarSemantics) {
// Maintain compatibility with the old version
if (currentTransaction != null) {
return Collections.singletonList(new PulsarSinkState(currentTransaction.getTxnID()));
}
}
return Collections.emptyList();
}Issue 10: Visibility issue with currentTransaction in concurrent scenariosLocation: private Transaction currentTransaction; // ← No volatileProblem Description:
Potential Risks:
Impact Scope:
Severity: MINOR (current framework assumes single-threading) Improvement Suggestions: private final AtomicReference<Transaction> currentTransaction = new AtomicReference<>(); |
Collaborator
|
Please enable CI following the instructions. |
davidzollo
reviewed
Mar 7, 2026
davidzollo
reviewed
Mar 7, 2026
Contributor
There was a problem hiding this comment.
The license statement cannot be deleted.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose of this pull request
This PR adds multi-table sink support for the Pulsar connector.
The Pulsar sink can now route records dynamically to different Pulsar topics
based on
SeaTunnelRow.getTableId(). Each topic maintains a dedicatedPulsar producer stored in a producer map, allowing the connector to support
multiple tables in a single pipeline.
This change implements
SupportMultiTableSinkfor the Pulsar connector.Closes #10426
Does this PR introduce any user-facing change?
Yes.
Previously, the Pulsar sink only supported writing records to a single topic.
With this change, the connector supports multi-table pipelines by routing
records to different Pulsar topics using
SeaTunnelRow.getTableId().How was this patch tested?
The change was tested locally by building the connector module and verifying
the multi-table routing logic.
Steps performed:
mvn clean installmvn spotless:checkThe routing behavior was reviewed to ensure records are written to different
Pulsar topics based on
SeaTunnelRow.getTableId().No additional unit tests were added because the change only affects routing
logic and existing Pulsar sink functionality remains unchanged.
Check list
incompatible-changes.mdconnector configuration files are updated (not applicable for this change)