This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/rocketmq-flink.git
The following commit(s) were added to refs/heads/main by this push:
new 7170770 [ISSUE #141] fix: commit transactions on the node holding the
context and close the remaining silent paths (#149)
7170770 is described below
commit 7170770e263688f403ff6e3e6cd1abe32117e836
Author: lizhimins <[email protected]>
AuthorDate: Tue Oct 27 14:28:17 2026 +0800
[ISSUE #141] fix: commit transactions on the node holding the context and
close the remaining silent paths (#149)
---
docs/remoting-connector.md | 8 ++
.../rocketmq/common/config/RocketMQOptions.java | 10 +-
.../connector/rocketmq/sink/InnerProducer.java | 14 ++
.../connector/rocketmq/sink/InnerProducerImpl.java | 76 +++++++---
.../rocketmq/sink/committer/SendCommittable.java | 19 +++
.../sink/committer/SendCommittableSerializer.java | 8 +-
.../rocketmq/sink/writer/RocketMQWriter.java | 51 +++++--
.../rocketmq/source/InnerConsumerImpl.java | 23 +++-
.../source/reader/RocketMQRecordEmitter.java | 56 ++++++--
.../source/reader/RocketMQSplitReader.java | 18 ++-
.../rocketmq/table/RocketMQDynamicTableSink.java | 8 --
.../table/RocketMQDynamicTableSinkFactory.java | 17 ++-
.../rocketmq/sink/InnerProducerImplTest.java | 153 +++++++++++++++++++++
.../committer/SendCommittableSerializerTest.java | 90 ++++++++++++
.../writer/RocketMQWriterQueueSelectorTest.java | 5 +
....java => RocketMQWriterSendPendingMaxTest.java} | 106 +++++++-------
... RocketMQWriterTransactionCommittableTest.java} | 101 ++++++--------
.../source/reader/RocketMQRecordEmitterTest.java | 90 +++++++++++-
.../reader/RocketMQSplitReaderPollTimeoutTest.java | 81 +++++++++++
.../table/RocketMQDynamicTableSinkFactoryTest.java | 60 +++++++-
.../RocketMQDynamicTableSinkMigrationTest.java | 2 -
21 files changed, 823 insertions(+), 173 deletions(-)
diff --git a/docs/remoting-connector.md b/docs/remoting-connector.md
index b31415a..c4c960a 100644
--- a/docs/remoting-connector.md
+++ b/docs/remoting-connector.md
@@ -50,6 +50,13 @@ RocketMQSink<String> sink = RocketMQSink.<String>builder()
- `AT_LEAST_ONCE` uses async sends with a pending counter and a flush barrier
on checkpoint.
- `EXACTLY_ONCE` uses `TransactionMQProducer` two-phase commit driven by Flink
checkpoints;
set `rocketmq.sink.transaction.timeout` larger than the checkpoint interval.
+- `EXACTLY_ONCE` also requires the **checkpoint interval to stay well below
the server-side
+ transaction state TTL**. A proxy keeps the transaction state of a half
message in its own memory
+ and drops it after `transactionDataExpireMillis` (30 seconds by default).
Flink only commits at
+ the end of a checkpoint, so the interval plus the checkpoint duration is the
delay that state has
+ to survive; a commit arriving after it is discarded while still being
answered with a success
+ code, so those records never appear and no error is raised. Raise the
server-side TTL if the job
+ needs longer checkpoints.
## Client options (`rocketmq.client.*`)
@@ -65,6 +72,7 @@ Shared by source and sink.
| `rocketmq.client.network.timeout.ms` | Long | 30000 | Client API timeout |
| `rocketmq.client.callback.threads` | Integer | CPU cores | Client callback
executor threads |
| `rocketmq.client.partition.discovery.interval.ms` | Long | 10000 |
Route/partition discovery interval |
+| `rocketmq.client.poll.timeout` | Long | 1000 | How long a source fetch
blocks while no record is available. Returns as soon as records arrive, so it
costs no data-path latency; it also bounds how long a fetch can outlive a close
request |
| `rocketmq.client.unitMode` / `rocketmq.client.unitName` | Boolean / String |
false / (none) | Unit mode routing |
| `rocketmq.client.debug` | Boolean | false | Verbose client logs |
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/common/config/RocketMQOptions.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/common/config/RocketMQOptions.java
index 9e7eef7..98d4de8 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/common/config/RocketMQOptions.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/common/config/RocketMQOptions.java
@@ -137,6 +137,12 @@ public class RocketMQOptions {
public static final ConfigOption<Long> POLL_TIMEOUT =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "poll.timeout")
.longType()
- .defaultValue(10L)
- .withDescription("how long to wait before giving up, the
unit is milliseconds");
+ .defaultValue(1000L)
+ .withDescription(
+ "How long a fetch blocks while no record is
available, in milliseconds."
+ + " The poll returns as soon as records
arrive, so this only"
+ + " bounds the idle wait and how long the
fetcher thread takes to"
+ + " notice a wakeup or a close request.
Keep it small enough for"
+ + " a timely shutdown and large enough to
stop an idle job from"
+ + " spinning the fetch loop.");
}
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducer.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducer.java
index aef6614..84f6614 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducer.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducer.java
@@ -73,6 +73,20 @@ public interface InnerProducer extends AutoCloseable {
*/
CompletableFuture<SendResult> sendMessageInTransaction(Message message);
+ /**
+ * Resolves the address this producer uses to reach the broker that
accepted the given send
+ * result, or null when the publish route does not know that broker.
+ *
+ * <p>The writer must call this right after a transactional send and hand
the address to the
+ * committer along with the {@link SendCommittable}: the node that
accepted the half message is
+ * the only one that can end the transaction, and a committer resolving
the address from its own
+ * route table may well pick a different node.
+ *
+ * @param sendResult the result of a send performed by this producer
+ * @return the address the send went to, or null if it cannot be resolved
+ */
+ String resolvePublishAddress(SendResult sendResult);
+
/**
* Commits the send operation identified by the specified SendCommittable
object.
*
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducerImpl.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducerImpl.java
index 6595f7a..156e259 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducerImpl.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/InnerProducerImpl.java
@@ -295,6 +295,31 @@ public class InnerProducerImpl implements InnerProducer {
Runnable::run);
}
+ @Override
+ public String resolvePublishAddress(final SendResult sendResult) {
+ return this.mqClientInstance.findBrokerAddressInPublish(
+ this.mqClientInstance.getBrokerNameFromMessageQueue(
+
producer.queueWithNamespace(sendResult.getMessageQueue())));
+ }
+
+ private String resolvePublishAddress(final String brokerName, final String
topic) {
+ String brokerAddress =
this.mqClientInstance.findBrokerAddressInPublish(brokerName);
+ if (brokerAddress == null) {
+ // The Committer owns a producer that never sends anything, so its
publish route table
+ // stays empty and findBrokerAddressInPublish() returns null.
invokeSync() then falls
+ // back to a name server channel, which cannot commit the half
message, and the reply is
+ // not a SUCCESS the caller would notice -- every record of the
checkpoint is lost
+ // without a single error. Refresh the route for the topic the
half message belongs to
+ // before resolving again.
+ LOG.warn(
+ "Publish route for {} is not loaded yet, refreshing before
ending transaction",
+ brokerName);
+ this.mqClientInstance.updateTopicRouteInfoFromNameServer(topic);
+ brokerAddress =
this.mqClientInstance.findBrokerAddressInPublish(brokerName);
+ }
+ return brokerAddress;
+ }
+
public void endTransaction(
final SendCommittable sendCommittable, final TransactionResult
transactionResult) {
@@ -302,20 +327,14 @@ public class InnerProducerImpl implements InnerProducer {
final String brokerName =
this.mqClientInstance.getBrokerNameFromMessageQueue(
producer.queueWithNamespace(sendCommittable.getMessageQueue()));
- String brokerAddress =
this.mqClientInstance.findBrokerAddressInPublish(brokerName);
- if (brokerAddress == null) {
- // The Committer owns a producer that never sends anything, so
its publish route
- // table stays empty and findBrokerAddressInPublish() returns
null. invokeSync()
- // then falls back to a name server channel, which cannot
commit the half message,
- // and the reply is not a SUCCESS the caller would notice --
every record of the
- // checkpoint is lost without a single error. Refresh the
route for the topic the
- // half message belongs to before resolving again.
- LOG.warn(
- "Publish route for {} is not loaded yet, refreshing
before ending transaction",
- brokerName);
- this.mqClientInstance.updateTopicRouteInfoFromNameServer(
- sendCommittable.getTopic());
- brokerAddress =
this.mqClientInstance.findBrokerAddressInPublish(brokerName);
+ // The node that accepted the half message is the only one holding
its transaction
+ // context, so prefer the address the writer recorded at send
time. Resolving it here
+ // from the committer's own route table can pick a different node,
and a commit sent to
+ // a node without the context is discarded while still being
answered with a success
+ // code -- the record is lost with nothing to show for it.
+ String brokerAddress = sendCommittable.getBrokerAddress();
+ if (StringUtils.isNullOrWhitespaceOnly(brokerAddress)) {
+ brokerAddress = resolvePublishAddress(brokerName,
sendCommittable.getTopic());
}
if (brokerAddress == null) {
throw new IllegalStateException(
@@ -413,16 +432,33 @@ public class InnerProducerImpl implements InnerProducer {
@Override
public CompletableFuture<Void> commit(SendCommittable sendCommittable) {
- return CompletableFuture.runAsync(
- () -> endTransaction(sendCommittable,
TransactionResult.COMMIT),
- producer.getExecutorService());
+ return endTransactionAsync(sendCommittable, TransactionResult.COMMIT);
}
@Override
public CompletableFuture<Void> rollback(SendCommittable sendCommittable) {
- return CompletableFuture.runAsync(
- () -> endTransaction(sendCommittable,
TransactionResult.ROLLBACK),
- producer.getExecutorService());
+ return endTransactionAsync(sendCommittable,
TransactionResult.ROLLBACK);
+ }
+
+ /**
+ * Ends the transaction on the calling thread. Submitting to the
producer's callback executor
+ * instead looked parallel but was lossy: that executor is bounded (4
threads, 2000 queued
+ * tasks, abort policy), so a checkpoint carrying more committables than
the queue holds
+ * rejected the surplus, and a rejected commit retried after the
server-side transaction context
+ * expired is silently dropped. The committer blocks until every request
of the checkpoint is
+ * done anyway, so running here turns a burst into visible checkpoint
duration instead of lost
+ * half messages.
+ */
+ private CompletableFuture<Void> endTransactionAsync(
+ final SendCommittable sendCommittable, final TransactionResult
transactionResult) {
+ try {
+ endTransaction(sendCommittable, transactionResult);
+ return CompletableFuture.completedFuture(null);
+ } catch (Exception e) {
+ CompletableFuture<Void> failed = new CompletableFuture<>();
+ failed.completeExceptionally(e);
+ return failed;
+ }
}
@Override
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/committer/SendCommittable.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/committer/SendCommittable.java
index a69e737..76de00a 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/committer/SendCommittable.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/committer/SendCommittable.java
@@ -42,6 +42,14 @@ public class SendCommittable {
private String transactionId;
+ /**
+ * Address of the node that accepted the half message, resolved from the
writer's own publish
+ * route right after the send. A proxy keeps the transaction context of a
half message in its
+ * own memory, so ending the transaction on any other node is discarded
while still being
+ * answered with a success code.
+ */
+ private String brokerAddress;
+
public SendCommittable() {}
public SendCommittable(SendResult sendResult) {
@@ -110,6 +118,14 @@ public class SendCommittable {
this.transactionId = transactionId;
}
+ public String getBrokerAddress() {
+ return brokerAddress;
+ }
+
+ public void setBrokerAddress(String brokerAddress) {
+ this.brokerAddress = brokerAddress;
+ }
+
public MessageQueue getMessageQueue() {
return new MessageQueue(topic, brokerName, queueId);
}
@@ -148,6 +164,9 @@ public class SendCommittable {
+ ", transactionId='"
+ transactionId
+ '\''
+ + ", brokerAddress='"
+ + brokerAddress
+ + '\''
+ '}';
}
}
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/committer/SendCommittableSerializer.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/committer/SendCommittableSerializer.java
index 4b64873..a44277f 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/committer/SendCommittableSerializer.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/committer/SendCommittableSerializer.java
@@ -28,7 +28,9 @@ import java.io.IOException;
/** A serializer used to serialize {@link SendCommittable}. */
public class SendCommittableSerializer implements
SimpleVersionedSerializer<SendCommittable> {
- private static final int CURRENT_VERSION = 1;
+ // Version 2 adds the address of the node that accepted the half message.
Committables written
+ // by version 1 are still readable: they simply carry no address and fall
back to route lookup.
+ private static final int CURRENT_VERSION = 2;
@Override
public int getVersion() {
@@ -46,6 +48,7 @@ public class SendCommittableSerializer implements
SimpleVersionedSerializer<Send
out.writeUTF(obj.getMsgId());
out.writeUTF(obj.getOffsetMsgId());
out.writeUTF(obj.getTransactionId());
+ out.writeUTF(obj.getBrokerAddress() != null ?
obj.getBrokerAddress() : "");
out.flush();
return stream.toByteArray();
}
@@ -63,6 +66,9 @@ public class SendCommittableSerializer implements
SimpleVersionedSerializer<Send
sendCommittable.setMsgId(in.readUTF());
sendCommittable.setOffsetMsgId(in.readUTF());
sendCommittable.setTransactionId(in.readUTF());
+ if (version >= 2) {
+ sendCommittable.setBrokerAddress(in.readUTF());
+ }
return sendCommittable;
}
}
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriter.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriter.java
index 59ad65c..6916a3d 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriter.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriter.java
@@ -48,6 +48,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@@ -62,10 +63,16 @@ public class RocketMQWriter<IN>
private final MessageQueueSelector messageQueueSelector;
private final RocketMQSinkContext rocketmqSinkContext;
private final RocketMQSerializationSchema<IN> serializationSchema;
- private final Map<String, SendResult> sendResultMap;
+ private final Map<String, SendCommittable> committableMap;
private final int sendRetryTimes;
private final long sendSleepTimeMs;
private final AtomicLong pendingRecords = new AtomicLong(0);
+ /**
+ * Bounds the in-flight async sends. {@code send.pending.max} was declared
with exactly this
+ * description but never read, so the send buffer grew without limit under
a slow broker.
+ */
+ private final Semaphore sendPermits;
+
private final AtomicReference<Throwable> asyncSendException = new
AtomicReference<>();
public RocketMQWriter(
@@ -91,10 +98,12 @@ public class RocketMQWriter<IN>
this.messageQueueSelector = messageQueueSelector;
this.serializationSchema = serializationSchema;
this.rocketmqSinkContext = new RocketMQSinkContextImpl(initContext,
configuration);
- this.sendResultMap = new ConcurrentHashMap<>();
+ this.committableMap = new ConcurrentHashMap<>();
this.sendRetryTimes =
configuration.getInteger(RocketMQSinkOptions.SEND_RETRY_TIMES);
this.sendSleepTimeMs =
configuration.getLong(RocketMQSinkOptions.OPTIONAL_WRITE_SLEEP_TIME_MS);
+ this.sendPermits =
+ new
Semaphore(configuration.getInteger(RocketMQSinkOptions.SEND_PENDING_MAX));
try {
serializationSchema.open(
initContext.asSerializationSchemaInitializationContext(),
rocketmqSinkContext);
@@ -121,19 +130,35 @@ public class RocketMQWriter<IN>
return;
}
+ try {
+ sendPermits.acquire();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while waiting for send
capacity", e);
+ }
pendingRecords.incrementAndGet();
// The element is handed to the selector as its sharding argument,
which is what lets a
// HashMessageQueueSelector route records that share a key to the same
queue.
- CompletableFuture<SendResult> sendFuture =
- messageQueueSelector == null
- ? producer.send(message)
- : producer.send(message, messageQueueSelector,
element);
+ CompletableFuture<SendResult> sendFuture;
+ try {
+ sendFuture =
+ messageQueueSelector == null
+ ? producer.send(message)
+ : producer.send(message, messageQueueSelector,
element);
+ } catch (RuntimeException | Error e) {
+ // A synchronously throwing send never reaches whenComplete, so
the capacity and the
+ // pending count have to be given back here or the writer would
wedge at the limit.
+ pendingRecords.decrementAndGet();
+ sendPermits.release();
+ throw e;
+ }
sendFuture.whenComplete(
(sendResult, throwable) -> {
if (throwable != null) {
asyncSendException.compareAndSet(null, throwable);
}
pendingRecords.decrementAndGet();
+ sendPermits.release();
});
}
@@ -143,7 +168,12 @@ public class RocketMQWriter<IN>
while (attempt <= sendRetryTimes) {
try {
SendResult sendResult =
producer.sendMessageInTransaction(message).get();
- sendResultMap.put(sendResult.getTransactionId(), sendResult);
+ SendCommittable committable = new SendCommittable(sendResult);
+ // Record the address the half message was actually accepted
by. Only that node
+ // holds the transaction context needed to end it, and the
committer owns a
+ // producer of its own whose route table may well resolve to a
different node.
+
committable.setBrokerAddress(producer.resolvePublishAddress(sendResult));
+ committableMap.put(sendResult.getTransactionId(), committable);
return; // success
} catch (Exception e) {
lastException = e;
@@ -192,12 +222,9 @@ public class RocketMQWriter<IN>
public Collection<SendCommittable> prepareCommit() throws IOException,
InterruptedException {
LOG.info("Prepare commit");
if (deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE) {
- final List<SendCommittable> committables = new ArrayList<>();
- for (SendResult sendResult : sendResultMap.values()) {
- committables.add(new SendCommittable(sendResult));
- }
+ final List<SendCommittable> committables = new
ArrayList<>(committableMap.values());
LOG.info("Committable size={}.", committables.size());
- sendResultMap.clear();
+ committableMap.clear();
return committables;
}
return Collections.emptyList();
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/InnerConsumerImpl.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/InnerConsumerImpl.java
index ef621cc..61d13f4 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/InnerConsumerImpl.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/InnerConsumerImpl.java
@@ -288,11 +288,12 @@ public class InnerConsumerImpl implements InnerConsumer {
@Override
public void wakeup() {
// Deliberately does not pause the assigned queues.
DefaultLitePullConsumer offers no
- // interruptible poll, but poll(Duration) is already bounded by a
short timeout, so the
- // fetch loop never blocks long enough to need forcing. Pausing here
without a matching
- // resume() would permanently stop delivering records while the client
keeps pulling and
- // advancing offsets in the background -- silent message loss with
nothing logged.
- // Use pause(Collection)/resume(Collection) as a pair if queue pausing
is ever needed.
+ // interruptible poll, but poll(Duration) is bounded by
rocketmq.client.poll.timeout, so the
+ // fetch loop comes back on its own within that bound instead of
needing to be forced.
+ // Pausing here without a matching resume() would permanently stop
delivering records while
+ // the client keeps pulling and advancing offsets in the background --
silent message loss
+ // with nothing logged. Use pause(Collection)/resume(Collection) as a
pair if queue pausing
+ // is ever needed.
LOG.debug(
"Wakeup requested; relying on the bounded poll timeout instead
of pausing queues");
}
@@ -463,7 +464,17 @@ public class InnerConsumerImpl implements InnerConsumer {
@Override
public CompletableFuture<Void> commitOffset(MessageQueue messageQueue,
long offset) {
return CompletableFuture.runAsync(
- () -> consumer.getOffsetStore().updateOffset(messageQueue,
offset, true),
+ () -> {
+ // The broker copy of the offset is display-only:
positioning always comes from
+ // Flink state. Publishing the checkpointed offset --
rollbacks included, hence
+ // increaseOnly=false -- keeps consumer progress and lag
monitoring truthful
+ // instead of leaving a stale high-water mark; the
previous code updated only
+ // the
+ // local offset store with increaseOnly=true and never
persisted, so it had no
+ // effect at all.
+ consumer.getOffsetStore().updateOffset(messageQueue,
offset, false);
+ consumer.getOffsetStore().persist(messageQueue);
+ },
commonExecutorService);
}
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQRecordEmitter.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQRecordEmitter.java
index 12afce2..f72151f 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQRecordEmitter.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQRecordEmitter.java
@@ -24,12 +24,17 @@ import
org.apache.flink.connector.rocketmq.source.reader.deserializer.RocketMQDe
import
org.apache.flink.connector.rocketmq.source.split.RocketMQSourceSplitState;
import org.apache.flink.util.Collector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.IOException;
/** The {@link RecordEmitter} implementation for {@link RocketMQSourceReader}.
*/
public class RocketMQRecordEmitter<T>
implements RecordEmitter<MessageView, T, RocketMQSourceSplitState> {
+ private static final Logger LOG =
LoggerFactory.getLogger(RocketMQRecordEmitter.class);
+
private final RocketMQDeserializationSchema<T> deserializationSchema;
private final SourceOutputWrapper<T> sourceOutputWrapper = new
SourceOutputWrapper<>();
@@ -44,18 +49,34 @@ public class RocketMQRecordEmitter<T>
sourceOutputWrapper.setSourceOutput(output);
sourceOutputWrapper.setTimestamp(element.getEventTime());
+ sourceOutputWrapper.resetCollectError();
try {
deserializationSchema.deserialize(element, sourceOutputWrapper);
} catch (Exception e) {
- // Naming the record makes a poison message identifiable from the
exception alone;
- // otherwise the only way to find it is to replay the whole queue.
- throw new IOException(
- String.format(
- "Failed to deserialize message %s from topic %s,
queue %d, offset %d",
- element.getMessageId(),
- element.getTopic(),
- element.getQueueId(),
- element.getQueueOffset()),
+ RuntimeException collectError =
sourceOutputWrapper.getCollectError();
+ if (collectError != null) {
+ // The record deserialized fine; the failure came from the
downstream sink while the
+ // already-deserialized record was being emitted (collect runs
the writer inline).
+ // Dropping it here would hide a real write failure behind a
"deserialization" warn.
+ throw new IOException(
+ String.format(
+ "Failed to emit message %s from topic %s,
queue %d, offset %d",
+ element.getMessageId(),
+ element.getTopic(),
+ element.getQueueId(),
+ element.getQueueOffset()),
+ collectError);
+ }
+ // A record that cannot be deserialized is dropped instead of
failing the task: with a
+ // throwing schema the job would otherwise loop "fail, restore,
re-read the same record"
+ // forever and rewrite everything it had already emitted. Naming
the record keeps the
+ // poison message identifiable from the log line alone.
+ LOG.warn(
+ "Dropped message {} from topic {}, queue {}, offset {}
because deserialization failed",
+ element.getMessageId(),
+ element.getTopic(),
+ element.getQueueId(),
+ element.getQueueOffset(),
e);
}
splitState.setCurrentOffset(element.getQueueOffset() + 1);
@@ -67,9 +88,16 @@ public class RocketMQRecordEmitter<T>
private long timestamp;
+ private RuntimeException collectError;
+
@Override
public void collect(T record) {
- sourceOutput.collect(record, timestamp);
+ try {
+ sourceOutput.collect(record, timestamp);
+ } catch (RuntimeException e) {
+ collectError = e;
+ throw e;
+ }
}
@Override
@@ -82,5 +110,13 @@ public class RocketMQRecordEmitter<T>
private void setTimestamp(long timestamp) {
this.timestamp = timestamp;
}
+
+ private void resetCollectError() {
+ collectError = null;
+ }
+
+ private RuntimeException getCollectError() {
+ return collectError;
+ }
}
}
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQSplitReader.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQSplitReader.java
index 17c7e82..294b2d4 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQSplitReader.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQSplitReader.java
@@ -19,6 +19,7 @@
package org.apache.flink.connector.rocketmq.source.reader;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
@@ -110,6 +111,21 @@ public class RocketMQSplitReader<T> implements
SplitReader<MessageView, RocketMQ
SourceReaderContext sourceReaderContext,
RocketMQDeserializationSchema<T> deserializationSchema,
RocketMQSourceReaderMetrics rocketmqSourceReaderMetrics) {
+ this(
+ configuration,
+ sourceReaderContext,
+ deserializationSchema,
+ rocketmqSourceReaderMetrics,
+ new InnerConsumerImpl(configuration));
+ }
+
+ @VisibleForTesting
+ RocketMQSplitReader(
+ Configuration configuration,
+ SourceReaderContext sourceReaderContext,
+ RocketMQDeserializationSchema<T> deserializationSchema,
+ RocketMQSourceReaderMetrics rocketmqSourceReaderMetrics,
+ InnerConsumer consumer) {
this.configuration = configuration;
this.sourceReaderContext = sourceReaderContext;
@@ -120,7 +136,7 @@ public class RocketMQSplitReader<T> implements
SplitReader<MessageView, RocketMQ
this.initFinish = false;
this.lock = new SpinLock();
- this.consumer = new InnerConsumerImpl(configuration);
+ this.consumer = consumer;
this.consumer.start();
this.rocketmqSourceReaderMetrics = rocketmqSourceReaderMetrics;
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSink.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSink.java
index 1a4bb04..a13146d 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSink.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSink.java
@@ -59,7 +59,6 @@ public class RocketMQDynamicTableSink implements
DynamicTableSink, SupportsWriti
private final String accessKey;
private final String secretKey;
- private final long retryTimes;
private final long sleepTime;
private final boolean isDynamicTag;
@@ -86,7 +85,6 @@ public class RocketMQDynamicTableSink implements
DynamicTableSink, SupportsWriti
String dynamicColumn,
String fieldDelimiter,
String encoding,
- long retryTimes,
long sleepTime,
boolean isDynamicTag,
boolean isDynamicTagIncluded,
@@ -110,7 +108,6 @@ public class RocketMQDynamicTableSink implements
DynamicTableSink, SupportsWriti
dynamicColumn,
fieldDelimiter,
encoding,
- retryTimes,
sleepTime,
isDynamicTag,
isDynamicTagIncluded,
@@ -135,7 +132,6 @@ public class RocketMQDynamicTableSink implements
DynamicTableSink, SupportsWriti
String dynamicColumn,
String fieldDelimiter,
String encoding,
- long retryTimes,
long sleepTime,
boolean isDynamicTag,
boolean isDynamicTagIncluded,
@@ -157,7 +153,6 @@ public class RocketMQDynamicTableSink implements
DynamicTableSink, SupportsWriti
this.dynamicColumn = dynamicColumn;
this.fieldDelimiter = fieldDelimiter;
this.encoding = encoding;
- this.retryTimes = retryTimes;
this.sleepTime = sleepTime;
this.isDynamicTag = isDynamicTag;
this.isDynamicTagIncluded = isDynamicTagIncluded;
@@ -208,8 +203,6 @@ public class RocketMQDynamicTableSink implements
DynamicTableSink, SupportsWriti
sinkConfig.setLong(RocketMQSinkOptions.SEND_TIMEOUT, sendTimeout);
sinkConfig.setInteger(RocketMQSinkOptions.SEND_PENDING_MAX,
sendPendingMax);
sinkConfig.setInteger(RocketMQSinkOptions.EXECUTOR_NUM, executorNum);
- // Pass legacy retry config that was previously stored but never
forwarded (P1 fix)
- sinkConfig.setInteger(RocketMQSinkOptions.OPTIONAL_WRITE_RETRY_TIMES,
(int) retryTimes);
sinkConfig.setLong(RocketMQSinkOptions.OPTIONAL_WRITE_SLEEP_TIME_MS,
sleepTime);
sinkBuilder.setConfig(sinkConfig);
@@ -244,7 +237,6 @@ public class RocketMQDynamicTableSink implements
DynamicTableSink, SupportsWriti
dynamicColumn,
fieldDelimiter,
encoding,
- retryTimes,
sleepTime,
isDynamicTag,
isDynamicTagIncluded,
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkFactory.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkFactory.java
index 3c5f240..2c11e92 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkFactory.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkFactory.java
@@ -111,7 +111,6 @@ public class RocketMQDynamicTableSinkFactory implements
DynamicTableSinkFactory
String dynamicColumn =
properties.getString(OPTIONAL_WRITE_DYNAMIC_TAG_COLUMN);
String encoding = properties.getString(OPTIONAL_ENCODING);
String fieldDelimiter = properties.getString(OPTIONAL_FIELD_DELIMITER);
- int retryTimes = properties.getInteger(OPTIONAL_WRITE_RETRY_TIMES);
long sleepTimeMs = properties.getLong(OPTIONAL_WRITE_SLEEP_TIME_MS);
boolean isDynamicTag =
properties.getBoolean(OPTIONAL_WRITE_IS_DYNAMIC_TAG);
boolean isDynamicTagIncluded =
@@ -123,7 +122,20 @@ public class RocketMQDynamicTableSinkFactory implements
DynamicTableSinkFactory
keyColumns = keyColumnsConfig.split(",");
}
String deliveryGuarantee =
properties.getString(RocketMQSinkOptions.DELIVERY_GUARANTEE);
- int sendRetryTimes =
properties.getInteger(RocketMQSinkOptions.SEND_RETRY_TIMES);
+ // `retry.times` is the legacy DDL name of the send retry count. The
writer only reads
+ // SEND_RETRY_TIMES, so honour the legacy key when the new one is
absent instead of
+ // silently dropping it; with neither set the documented default (3)
applies.
+ int sendRetryTimes =
+ properties
+ .getOptional(RocketMQSinkOptions.SEND_RETRY_TIMES)
+ .orElseGet(
+ () ->
+ properties
+
.getOptional(OPTIONAL_WRITE_RETRY_TIMES)
+ .orElseGet(
+ () ->
+
RocketMQSinkOptions.SEND_RETRY_TIMES
+
.defaultValue()));
long sendTimeout =
properties.getLong(RocketMQSinkOptions.SEND_TIMEOUT);
int sendPendingMax =
properties.getInteger(RocketMQSinkOptions.SEND_PENDING_MAX);
int executorNum =
properties.getInteger(RocketMQSinkOptions.EXECUTOR_NUM);
@@ -144,7 +156,6 @@ public class RocketMQDynamicTableSinkFactory implements
DynamicTableSinkFactory
dynamicColumn,
fieldDelimiter,
encoding,
- retryTimes,
sleepTimeMs,
isDynamicTag,
isDynamicTagIncluded,
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/InnerProducerImplTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/InnerProducerImplTest.java
index 84a216e..7c0b016 100644
---
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/InnerProducerImplTest.java
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/InnerProducerImplTest.java
@@ -18,11 +18,23 @@
package org.apache.flink.connector.rocketmq.sink;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.rocketmq.sink.committer.SendCommittable;
+import org.apache.rocketmq.client.impl.MQClientAPIImpl;
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
+import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
+import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
@@ -87,4 +99,145 @@ class InnerProducerImplTest {
assertThat(producer.getRetryTimesWhenSendAsyncFailed()).isEqualTo(3);
}
}
+
+ /**
+ * The node that accepted the half message is the only one holding its
transaction context, so
+ * the commit must go to the address the writer recorded. Resolving it
from the committer's own
+ * route table can name another node, which discards the commit and still
answers with a success
+ * code.
+ */
+ @Test
+ void commitGoesToTheAddressRecordedByTheWriterTest() throws Exception {
+ MQClientInstance clientInstance = Mockito.mock(MQClientInstance.class);
+ NettyRemotingClient remotingClient = stubRemoting(clientInstance);
+
+ try (InnerProducerImpl impl = new InnerProducerImpl(configuration())) {
+ setClientInstance(impl, clientInstance);
+
+ impl.endTransaction(committable("10.0.38.29:8080"),
TransactionResult.COMMIT);
+
+ Mockito.verify(remotingClient)
+ .invokeSync(
+ Mockito.eq("10.0.38.29:8080"),
+ Mockito.any(RemotingCommand.class),
+ Mockito.anyLong());
+ Mockito.verify(clientInstance, Mockito.never())
+ .findBrokerAddressInPublish(Mockito.anyString());
+ }
+ }
+
+ /** Committables restored from state written before the address was
recorded have no address. */
+ @Test
+ void commitFallsBackToThePublishRouteWithoutRecordedAddressTest() throws
Exception {
+ MQClientInstance clientInstance = Mockito.mock(MQClientInstance.class);
+
Mockito.when(clientInstance.getBrokerNameFromMessageQueue(Mockito.any()))
+ .thenReturn("broker-a");
+ Mockito.when(clientInstance.findBrokerAddressInPublish("broker-a"))
+ .thenReturn("10.0.26.60:8080");
+ NettyRemotingClient remotingClient = stubRemoting(clientInstance);
+
+ try (InnerProducerImpl impl = new InnerProducerImpl(configuration())) {
+ setClientInstance(impl, clientInstance);
+
+ impl.endTransaction(committable(null), TransactionResult.COMMIT);
+
+ Mockito.verify(remotingClient)
+ .invokeSync(
+ Mockito.eq("10.0.26.60:8080"),
+ Mockito.any(RemotingCommand.class),
+ Mockito.anyLong());
+ }
+ }
+
+ /**
+ * A checkpoint can carry thousands of committables; committing them must
not depend on a
+ * bounded executor queue that rejects the surplus (a rejected commit is a
lost commit once the
+ * server-side transaction context expires).
+ */
+ @Test
+ void commitBurstShouldNotRejectTest() throws Exception {
+ MQClientInstance clientInstance = Mockito.mock(MQClientInstance.class);
+ NettyRemotingClient remotingClient = stubRemoting(clientInstance);
+
+ try (InnerProducerImpl impl = new InnerProducerImpl(configuration())) {
+ setClientInstance(impl, clientInstance);
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ for (int i = 0; i < 3000; i++) {
+ futures.add(impl.commit(committable("10.0.26.60:8080")));
+ }
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
+ .get(30, TimeUnit.SECONDS);
+ Mockito.verify(remotingClient, Mockito.times(3000))
+ .invokeSync(
+ Mockito.eq("10.0.26.60:8080"),
+ Mockito.any(RemotingCommand.class),
+ Mockito.anyLong());
+ }
+ }
+
+ @Test
+ void commitFailureShouldCompleteTheFutureExceptionallyTest() throws
Exception {
+ MQClientInstance clientInstance = Mockito.mock(MQClientInstance.class);
+ MQClientAPIImpl api = Mockito.mock(MQClientAPIImpl.class);
+ NettyRemotingClient remotingClient =
Mockito.mock(NettyRemotingClient.class);
+ Mockito.when(clientInstance.getMQClientAPIImpl()).thenReturn(api);
+ Mockito.when(api.getRemotingClient()).thenReturn(remotingClient);
+ Mockito.when(
+ remotingClient.invokeSync(
+ Mockito.anyString(),
+ Mockito.any(RemotingCommand.class),
+ Mockito.anyLong()))
+ .thenThrow(new
RemotingSendRequestException("10.0.26.60:8080"));
+
+ try (InnerProducerImpl impl = new InnerProducerImpl(configuration())) {
+ setClientInstance(impl, clientInstance);
+
+ CompletableFuture<Void> future =
impl.commit(committable("10.0.26.60:8080"));
+ assertThat(future).isCompletedExceptionally();
+ }
+ }
+
+ private static Configuration configuration() {
+ Configuration config = new Configuration();
+ config.setString(RocketMQSinkOptions.ENDPOINTS, "127.0.0.1:9876");
+ config.setString(RocketMQSinkOptions.PRODUCER_GROUP, "test-group");
+ return config;
+ }
+
+ private static NettyRemotingClient stubRemoting(MQClientInstance
clientInstance)
+ throws Exception {
+ MQClientAPIImpl api = Mockito.mock(MQClientAPIImpl.class);
+ NettyRemotingClient remotingClient =
Mockito.mock(NettyRemotingClient.class);
+ Mockito.when(clientInstance.getMQClientAPIImpl()).thenReturn(api);
+ Mockito.when(api.getRemotingClient()).thenReturn(remotingClient);
+ Mockito.when(
+ remotingClient.invokeSync(
+ Mockito.anyString(),
+ Mockito.any(RemotingCommand.class),
+ Mockito.anyLong()))
+
.thenReturn(RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, ""));
+ return remotingClient;
+ }
+
+ private static void setClientInstance(InnerProducerImpl impl,
MQClientInstance instance)
+ throws Exception {
+ Field field =
InnerProducerImpl.class.getDeclaredField("mqClientInstance");
+ field.setAccessible(true);
+ field.set(impl, instance);
+ }
+
+ private static SendCommittable committable(String brokerAddress) {
+ SendCommittable committable = new SendCommittable();
+ committable.setTopic("TopicTest");
+ committable.setBrokerName("broker-a");
+ committable.setQueueId(3);
+ committable.setQueueOffset(42L);
+ committable.setMsgId("msg-1");
+ // 32 hex chars -- 8 ip, 8 port, 16 commit log offset -- as decoded by
MessageDecoder.
+ committable.setOffsetMsgId("0A00000100002382000000000000002A");
+ committable.setTransactionId("trans-1");
+ committable.setBrokerAddress(brokerAddress);
+ return committable;
+ }
}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/committer/SendCommittableSerializerTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/committer/SendCommittableSerializerTest.java
new file mode 100644
index 0000000..a4171e9
--- /dev/null
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/committer/SendCommittableSerializerTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.flink.connector.rocketmq.sink.committer;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link SendCommittableSerializer}. The address of the node that
accepted the half
+ * message travels with the committable because it survives a checkpoint;
losing it on the way would
+ * silently send every commit to whatever node the committer's route table
happens to name.
+ */
+class SendCommittableSerializerTest {
+
+ private final SendCommittableSerializer serializer = new
SendCommittableSerializer();
+
+ @Test
+ void brokerAddressSurvivesRoundTripTest() throws Exception {
+ SendCommittable committable = sendCommittable();
+ committable.setBrokerAddress("10.0.38.29:8080");
+
+ SendCommittable restored =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(committable));
+
+ assertThat(restored.getBrokerAddress()).isEqualTo("10.0.38.29:8080");
+ assertThat(restored.getTopic()).isEqualTo(committable.getTopic());
+
assertThat(restored.getBrokerName()).isEqualTo(committable.getBrokerName());
+ assertThat(restored.getQueueId()).isEqualTo(committable.getQueueId());
+
assertThat(restored.getQueueOffset()).isEqualTo(committable.getQueueOffset());
+ assertThat(restored.getMsgId()).isEqualTo(committable.getMsgId());
+
assertThat(restored.getOffsetMsgId()).isEqualTo(committable.getOffsetMsgId());
+
assertThat(restored.getTransactionId()).isEqualTo(committable.getTransactionId());
+ }
+
+ @Test
+ void versionOnePayloadIsStillReadableTest() throws Exception {
+ SendCommittable restored = serializer.deserialize(1,
versionOneBytes());
+
+ assertThat(restored.getTransactionId()).isEqualTo("trans-1");
+ assertThat(restored.getBrokerAddress()).isNull();
+ }
+
+ /** The version 1 layout: the same fields, without the trailing address. */
+ private static byte[] versionOneBytes() throws Exception {
+ ByteArrayOutputStream stream = new ByteArrayOutputStream();
+ try (DataOutputStream out = new DataOutputStream(stream)) {
+ out.writeUTF("TopicTest");
+ out.writeUTF("broker-a");
+ out.writeInt(3);
+ out.writeLong(42L);
+ out.writeUTF("msg-1");
+ out.writeUTF("offset-msg-1");
+ out.writeUTF("trans-1");
+ out.flush();
+ }
+ return stream.toByteArray();
+ }
+
+ private static SendCommittable sendCommittable() {
+ SendCommittable committable = new SendCommittable();
+ committable.setTopic("TopicTest");
+ committable.setBrokerName("broker-a");
+ committable.setQueueId(3);
+ committable.setQueueOffset(42L);
+ committable.setMsgId("msg-1");
+ committable.setOffsetMsgId("offset-msg-1");
+ committable.setTransactionId("trans-1");
+ return committable;
+ }
+}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
index 1828201..2bce6db 100644
---
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
@@ -130,6 +130,11 @@ class RocketMQWriterQueueSelectorTest {
throw new UnsupportedOperationException();
}
+ @Override
+ public String resolvePublishAddress(SendResult sendResult) {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public CompletableFuture<Void> commit(SendCommittable sendCommittable)
{
throw new UnsupportedOperationException();
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterSendPendingMaxTest.java
similarity index 58%
copy from
flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
copy to
flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterSendPendingMaxTest.java
index 1828201..9eac3bc 100644
---
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterSendPendingMaxTest.java
@@ -26,12 +26,12 @@ import
org.apache.flink.connector.rocketmq.sink.RocketMQSinkOptions;
import org.apache.flink.connector.rocketmq.sink.committer.SendCommittable;
import
org.apache.flink.connector.rocketmq.sink.writer.context.RocketMQSinkContext;
import
org.apache.flink.connector.rocketmq.sink.writer.serializer.RocketMQSerializationSchema;
-import
org.apache.flink.streaming.connectors.rocketmq.common.selector.HashMessageQueueSelector;
import
org.apache.flink.streaming.connectors.rocketmq.common.selector.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import org.mockito.Mockito;
import java.nio.charset.StandardCharsets;
@@ -42,46 +42,54 @@ import java.util.concurrent.CompletableFuture;
import static org.assertj.core.api.Assertions.assertThat;
/**
- * Tests that a configured {@link MessageQueueSelector} actually reaches the
producer. Accepting the
- * selector on the builder without ever passing it to the send call silently
turns ordered writes
- * into round robin writes, which no amount of configuration can reveal at
runtime.
+ * Tests that {@code send.pending.max} bounds the in-flight async sends. The
option ships a
+ * description promising exactly that ("Maximum number of pending messages in
the send buffer") but
+ * was never read, so a slow broker let the buffer grow without limit.
*/
-class RocketMQWriterQueueSelectorTest {
+class RocketMQWriterSendPendingMaxTest {
- private static final String TOPIC = "TopicWriterSelector";
+ private static final String TOPIC = "TopicWriterPending";
@Test
- void configuredSelectorReachesTheProducerTest() throws Exception {
- RecordingProducer producer = new RecordingProducer();
- MessageQueueSelector selector = new HashMessageQueueSelector();
-
- newWriter(selector, producer).write("sharding-key", null);
-
- assertThat(producer.plainSends).isEmpty();
- assertThat(producer.selectorSends).hasSize(1);
- assertThat(producer.selectorSends.get(0).selector).isSameAs(selector);
-
assertThat(producer.selectorSends.get(0).arg).isEqualTo("sharding-key");
-
assertThat(producer.selectorSends.get(0).message.getTopic()).isEqualTo(TOPIC);
+ @Timeout(30)
+ void writesBlockOnceThePendingLimitIsReachedTest() throws Exception {
+ GatedProducer producer = new GatedProducer();
+ RocketMQWriter<String> writer = newWriter(producer, 1);
+
+ writer.write("first", null);
+ assertThat(producer.sends).hasSize(1);
+
+ Thread second = new Thread(() -> uncheckedWrite(writer, "second"));
+ second.start();
+ second.join(1000);
+ assertThat(second.isAlive())
+ .as("the second write must block while the first send is still
in flight")
+ .isTrue();
+ assertThat(producer.sends).hasSize(1);
+
+ producer.completeAll();
+ second.join(5000);
+ assertThat(second.isAlive()).as("releasing the send must unblock the
writer").isFalse();
+ assertThat(producer.sends).hasSize(2);
+ writer.close();
}
- @Test
- void withoutSelectorThePlainSendIsUsedTest() throws Exception {
- RecordingProducer producer = new RecordingProducer();
-
- newWriter(null, producer).write("sharding-key", null);
-
- assertThat(producer.plainSends).hasSize(1);
- assertThat(producer.selectorSends).isEmpty();
+ private static void uncheckedWrite(RocketMQWriter<String> writer, String
record) {
+ try {
+ writer.write(record, null);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
}
- private static RocketMQWriter<String> newWriter(
- MessageQueueSelector selector, InnerProducer producer) {
+ private static RocketMQWriter<String> newWriter(GatedProducer producer,
int pendingMax) {
Configuration configuration = new Configuration();
configuration.setString(
RocketMQSinkOptions.DELIVERY_GUARANTEE,
DeliveryGuarantee.AT_LEAST_ONCE.name());
+ configuration.setInteger(RocketMQSinkOptions.SEND_PENDING_MAX,
pendingMax);
return new RocketMQWriter<>(
configuration,
- selector,
+ null,
new StringToMessageSchema(),
Mockito.mock(Sink.InitContext.class),
producer);
@@ -98,31 +106,30 @@ class RocketMQWriterQueueSelectorTest {
}
}
- /** Records which send overload the writer chose. */
- private static class RecordingProducer implements InnerProducer {
+ /** Keeps every send future open until the test releases them. */
+ private static class GatedProducer implements InnerProducer {
- private final List<Message> plainSends = new ArrayList<>();
- private final List<SelectorSend> selectorSends = new ArrayList<>();
+ private final List<CompletableFuture<SendResult>> sends = new
ArrayList<>();
@Override
public void start() {}
@Override
public String getProducerGroup() {
- return "recording-producer";
+ return "gated-producer";
}
@Override
- public CompletableFuture<SendResult> send(Message message) {
- plainSends.add(message);
- return CompletableFuture.completedFuture(new SendResult());
+ public synchronized CompletableFuture<SendResult> send(Message
message) {
+ CompletableFuture<SendResult> future = new CompletableFuture<>();
+ sends.add(future);
+ return future;
}
@Override
public CompletableFuture<SendResult> send(
Message message, MessageQueueSelector selector, Object arg) {
- selectorSends.add(new SelectorSend(message, selector, arg));
- return CompletableFuture.completedFuture(new SendResult());
+ return send(message);
}
@Override
@@ -131,28 +138,27 @@ class RocketMQWriterQueueSelectorTest {
}
@Override
- public CompletableFuture<Void> commit(SendCommittable sendCommittable)
{
+ public String resolvePublishAddress(SendResult sendResult) {
throw new UnsupportedOperationException();
}
@Override
- public CompletableFuture<Void> rollback(SendCommittable
sendCommittable) {
+ public CompletableFuture<Void> commit(SendCommittable sendCommittable)
{
throw new UnsupportedOperationException();
}
@Override
- public void close() {}
-
- private static class SelectorSend {
- private final Message message;
- private final MessageQueueSelector selector;
- private final Object arg;
+ public CompletableFuture<Void> rollback(SendCommittable
sendCommittable) {
+ throw new UnsupportedOperationException();
+ }
- private SelectorSend(Message message, MessageQueueSelector
selector, Object arg) {
- this.message = message;
- this.selector = selector;
- this.arg = arg;
+ synchronized void completeAll() {
+ for (CompletableFuture<SendResult> future : sends) {
+ future.complete(new SendResult());
}
}
+
+ @Override
+ public void close() {}
}
}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterTransactionCommittableTest.java
similarity index 57%
copy from
flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
copy to
flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterTransactionCommittableTest.java
index 1828201..0923131 100644
---
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterQueueSelectorTest.java
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/sink/writer/RocketMQWriterTransactionCommittableTest.java
@@ -26,62 +26,56 @@ import
org.apache.flink.connector.rocketmq.sink.RocketMQSinkOptions;
import org.apache.flink.connector.rocketmq.sink.committer.SendCommittable;
import
org.apache.flink.connector.rocketmq.sink.writer.context.RocketMQSinkContext;
import
org.apache.flink.connector.rocketmq.sink.writer.serializer.RocketMQSerializationSchema;
-import
org.apache.flink.streaming.connectors.rocketmq.common.selector.HashMessageQueueSelector;
import
org.apache.flink.streaming.connectors.rocketmq.common.selector.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageQueue;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import static org.assertj.core.api.Assertions.assertThat;
/**
- * Tests that a configured {@link MessageQueueSelector} actually reaches the
producer. Accepting the
- * selector on the builder without ever passing it to the send call silently
turns ordered writes
- * into round robin writes, which no amount of configuration can reveal at
runtime.
+ * Tests that a transactional write records the address the half message was
accepted by. The
+ * committer runs on a producer of its own, whose publish route is resolved
independently and may
+ * name a different node; a commit sent there is discarded while still
answered with a success code.
*/
-class RocketMQWriterQueueSelectorTest {
+class RocketMQWriterTransactionCommittableTest {
- private static final String TOPIC = "TopicWriterSelector";
+ private static final String TOPIC = "TopicWriterTransaction";
+ private static final String SEND_ADDRESS = "10.0.38.29:8080";
@Test
- void configuredSelectorReachesTheProducerTest() throws Exception {
- RecordingProducer producer = new RecordingProducer();
- MessageQueueSelector selector = new HashMessageQueueSelector();
-
- newWriter(selector, producer).write("sharding-key", null);
-
- assertThat(producer.plainSends).isEmpty();
- assertThat(producer.selectorSends).hasSize(1);
- assertThat(producer.selectorSends.get(0).selector).isSameAs(selector);
-
assertThat(producer.selectorSends.get(0).arg).isEqualTo("sharding-key");
-
assertThat(producer.selectorSends.get(0).message.getTopic()).isEqualTo(TOPIC);
+ void prepareCommitCarriesTheAddressOfTheSendingNodeTest() throws Exception
{
+ TransactionalProducer producer = new TransactionalProducer();
+ RocketMQWriter<String> writer = newWriter(producer);
+
+ writer.write("record", null);
+ Collection<SendCommittable> committables = writer.prepareCommit();
+
+ assertThat(committables).hasSize(1);
+ SendCommittable committable = committables.iterator().next();
+ assertThat(committable.getBrokerAddress()).isEqualTo(SEND_ADDRESS);
+ assertThat(committable.getTransactionId()).isEqualTo("trans-1");
+ assertThat(committable.getBrokerName()).isEqualTo("broker-a");
+ // The map must be drained, otherwise the next checkpoint commits the
same half message
+ // twice.
+ assertThat(writer.prepareCommit()).isEmpty();
}
- @Test
- void withoutSelectorThePlainSendIsUsedTest() throws Exception {
- RecordingProducer producer = new RecordingProducer();
-
- newWriter(null, producer).write("sharding-key", null);
-
- assertThat(producer.plainSends).hasSize(1);
- assertThat(producer.selectorSends).isEmpty();
- }
-
- private static RocketMQWriter<String> newWriter(
- MessageQueueSelector selector, InnerProducer producer) {
+ private static RocketMQWriter<String> newWriter(InnerProducer producer) {
Configuration configuration = new Configuration();
configuration.setString(
- RocketMQSinkOptions.DELIVERY_GUARANTEE,
DeliveryGuarantee.AT_LEAST_ONCE.name());
+ RocketMQSinkOptions.DELIVERY_GUARANTEE,
DeliveryGuarantee.EXACTLY_ONCE.name());
return new RocketMQWriter<>(
configuration,
- selector,
+ null,
new StringToMessageSchema(),
Mockito.mock(Sink.InitContext.class),
producer);
@@ -98,36 +92,43 @@ class RocketMQWriterQueueSelectorTest {
}
}
- /** Records which send overload the writer chose. */
- private static class RecordingProducer implements InnerProducer {
-
- private final List<Message> plainSends = new ArrayList<>();
- private final List<SelectorSend> selectorSends = new ArrayList<>();
+ /** Returns a send result plus the address the writer is expected to
record for it. */
+ private static class TransactionalProducer implements InnerProducer {
@Override
public void start() {}
@Override
public String getProducerGroup() {
- return "recording-producer";
+ return "transactional-producer";
}
@Override
public CompletableFuture<SendResult> send(Message message) {
- plainSends.add(message);
- return CompletableFuture.completedFuture(new SendResult());
+ throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<SendResult> send(
Message message, MessageQueueSelector selector, Object arg) {
- selectorSends.add(new SelectorSend(message, selector, arg));
- return CompletableFuture.completedFuture(new SendResult());
+ throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<SendResult> sendMessageInTransaction(Message
message) {
- throw new UnsupportedOperationException();
+ SendResult sendResult = new SendResult();
+ sendResult.setSendStatus(SendStatus.SEND_OK);
+ sendResult.setMessageQueue(new MessageQueue(TOPIC, "broker-a", 3));
+ sendResult.setQueueOffset(42L);
+ sendResult.setMsgId("msg-1");
+ sendResult.setOffsetMsgId("0A00000100002382000000000000002A");
+ sendResult.setTransactionId("trans-1");
+ return CompletableFuture.completedFuture(sendResult);
+ }
+
+ @Override
+ public String resolvePublishAddress(SendResult sendResult) {
+ return SEND_ADDRESS;
}
@Override
@@ -142,17 +143,5 @@ class RocketMQWriterQueueSelectorTest {
@Override
public void close() {}
-
- private static class SelectorSend {
- private final Message message;
- private final MessageQueueSelector selector;
- private final Object arg;
-
- private SelectorSend(Message message, MessageQueueSelector
selector, Object arg) {
- this.message = message;
- this.selector = selector;
- this.arg = arg;
- }
- }
}
}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQRecordEmitterTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQRecordEmitterTest.java
index 6a357e3..a9b08de 100644
---
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQRecordEmitterTest.java
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQRecordEmitterTest.java
@@ -26,6 +26,7 @@ import
org.apache.flink.connector.rocketmq.source.split.RocketMQSourceSplit;
import
org.apache.flink.connector.rocketmq.source.split.RocketMQSourceSplitState;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.Collector;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;
@@ -33,6 +34,10 @@ import org.junit.Test;
import java.io.IOException;
import java.net.InetSocketAddress;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
/** Test for {@link RocketMQRecordEmitter}. */
public class RocketMQRecordEmitterTest {
@@ -65,7 +70,90 @@ public class RocketMQRecordEmitterTest {
recordEmitter.emitRecord(messageView, new TestingEmitterOutput<>(),
partitionSplitState);
}
- private static final class TestingEmitterOutput<E> implements
ReaderOutput<E> {
+ /**
+ * A record whose deserialization throws must be dropped with a warning
instead of failing the
+ * task: otherwise the job loops "fail, restore, re-read the same record"
and rewrites
+ * everything it already emitted. The split offset still has to advance
past the poison record.
+ */
+ @Test
+ public void testUndeserializableRecordIsDroppedAndOffsetAdvances() throws
IOException {
+ RocketMQRecordEmitter<RowData> recordEmitter =
+ new RocketMQRecordEmitter<>(new ThrowingSchemaWrapper<>());
+ MessageExt message =
+ new MessageExt(
+ 1,
+ System.currentTimeMillis(),
+ InetSocketAddress.createUnresolved("127.0.0.1", 8080),
+ System.currentTimeMillis(),
+ InetSocketAddress.createUnresolved("127.0.0.1", 8088),
+ "184019388");
+ message.setBody("poison".getBytes());
+ message.setQueueOffset(100L);
+
+ RocketMQSourceSplitState splitState =
+ new RocketMQSourceSplitState(new RocketMQSourceSplit("t", "b",
0, 100L, -1L));
+
+ recordEmitter.emitRecord(
+ new MessageViewExt(message), new TestingEmitterOutput<>(),
splitState);
+
+ assertEquals(101L, splitState.getCurrentOffset());
+ }
+
+ /**
+ * A failure raised while emitting an already-deserialized record (the
downstream sink runs
+ * inline inside collect) must still fail the task: dropping it would hide
a real write failure.
+ */
+ @Test
+ public void testSinkFailureDuringCollectIsNotSwallowed() {
+ RocketMQRecordEmitter<RowData> recordEmitter =
+ new RocketMQRecordEmitter<>(new CollectingSchemaWrapper<>());
+ MessageExt message =
+ new MessageExt(
+ 1,
+ System.currentTimeMillis(),
+ InetSocketAddress.createUnresolved("127.0.0.1", 8080),
+ System.currentTimeMillis(),
+ InetSocketAddress.createUnresolved("127.0.0.1", 8088),
+ "184019389");
+ message.setBody("fine".getBytes());
+ message.setQueueOffset(7L);
+
+ RocketMQSourceSplitState splitState =
+ new RocketMQSourceSplitState(new RocketMQSourceSplit("t", "b",
0, 7L, -1L));
+
+ try {
+ recordEmitter.emitRecord(
+ new MessageViewExt(message), new FailingEmitterOutput<>(),
splitState);
+ fail("a sink failure during collect must propagate");
+ } catch (IOException expected) {
+ assertTrue(expected.getMessage().contains("Failed to emit
message"));
+ }
+ }
+
+ /** Emits exactly one record, so the output's failure happens inside
collect. */
+ private static final class CollectingSchemaWrapper<T> extends
RocketMQSchemaWrapper<T> {
+ @Override
+ public void deserialize(MessageView messageView, Collector<T> out) {
+ out.collect(null);
+ }
+ }
+
+ private static final class FailingEmitterOutput<E> extends
TestingEmitterOutput<E> {
+ @Override
+ public void collect(E record) {
+ throw new RuntimeException("sink rejected the record");
+ }
+ }
+
+ /** Deserialization always fails, as a schema rejecting an unparsable body
would. */
+ private static final class ThrowingSchemaWrapper<T> extends
RocketMQSchemaWrapper<T> {
+ @Override
+ public void deserialize(MessageView messageView, Collector<T> out)
throws IOException {
+ throw new IOException("deliberately unparsable body");
+ }
+ }
+
+ private static class TestingEmitterOutput<E> implements ReaderOutput<E> {
private TestingEmitterOutput() {}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQSplitReaderPollTimeoutTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQSplitReaderPollTimeoutTest.java
new file mode 100644
index 0000000..f8c38ca
--- /dev/null
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQSplitReaderPollTimeoutTest.java
@@ -0,0 +1,81 @@
+/*
+ * 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.flink.connector.rocketmq.source.reader;
+
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.rocketmq.common.config.RocketMQOptions;
+import org.apache.flink.connector.rocketmq.source.InnerConsumer;
+import
org.apache.flink.connector.rocketmq.source.metrics.RocketMQSourceReaderMetrics;
+import
org.apache.flink.connector.rocketmq.source.reader.deserializer.RocketMQDeserializationSchema;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests that {@code rocketmq.client.poll.timeout} reaches the consumer poll
and that its default no
+ * longer spins the fetch loop of an idle job a hundred times a second. The
bound matters because
+ * {@link RocketMQSplitReader#wakeUp()} only sets a flag and the consumer
deliberately has no
+ * interruptible poll, so the timeout is what limits how long a fetch can
outlive a wakeup.
+ */
+class RocketMQSplitReaderPollTimeoutTest {
+
+ @Test
+ void idleFetchBlocksForTheDefaultTimeoutTest() {
+ assertThat(pollTimeoutOf(new
Configuration())).isEqualTo(Duration.ofSeconds(1));
+ }
+
+ @Test
+ void idleFetchBlocksForTheConfiguredTimeoutTest() {
+ Configuration configuration = new Configuration();
+ configuration.set(RocketMQOptions.POLL_TIMEOUT, 250L);
+
+
assertThat(pollTimeoutOf(configuration)).isEqualTo(Duration.ofMillis(250));
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Duration pollTimeoutOf(Configuration configuration) {
+ InnerConsumer consumer = Mockito.mock(InnerConsumer.class);
+
Mockito.when(consumer.poll(Mockito.any())).thenReturn(Collections.emptyList());
+
+ RocketMQSplitReader<String> reader =
+ new RocketMQSplitReader<>(
+ configuration,
+ Mockito.mock(SourceReaderContext.class),
+ Mockito.mock(RocketMQDeserializationSchema.class),
+ Mockito.mock(RocketMQSourceReaderMetrics.class),
+ consumer);
+ try {
+ reader.fetch();
+ } catch (IOException e) {
+ throw new AssertionError("fetching without records must not fail",
e);
+ }
+
+ ArgumentCaptor<Duration> timeout =
ArgumentCaptor.forClass(Duration.class);
+ Mockito.verify(consumer).poll(timeout.capture());
+ return timeout.getValue();
+ }
+}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkFactoryTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkFactoryTest.java
index a06c8af..767b47a 100644
---
a/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkFactoryTest.java
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkFactoryTest.java
@@ -19,6 +19,9 @@
package org.apache.flink.streaming.connectors.rocketmq.table;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.rocketmq.common.config.RocketMQOptions;
+import org.apache.flink.connector.rocketmq.sink.RocketMQSink;
+import org.apache.flink.connector.rocketmq.sink.RocketMQSinkOptions;
import org.apache.flink.connector.rocketmq.source.RocketMQSourceOptions;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.ValidationException;
@@ -28,17 +31,21 @@ import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ResolvedCatalogTable;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
import org.apache.flink.table.factories.FactoryUtil;
+import
org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext;
import org.junit.Ignore;
import org.junit.Test;
+import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -62,7 +69,7 @@ public class RocketMQDynamicTableSinkFactoryTest {
public void testRocketMQDynamicTableSinkWithLegalOption() {
final Map<String, String> options = new HashMap<>();
options.put("connector", IDENTIFIER);
- options.put(RocketMQSourceOptions.TOPIC.key(), TOPIC);
+ options.put(RocketMQSinkOptions.TOPIC.key(), TOPIC);
options.put(
RocketMQSourceOptions.OPTIONAL_STARTUP_OFFSET_TIMESTAMP.key(),
String.valueOf(System.currentTimeMillis()));
@@ -94,6 +101,57 @@ public class RocketMQDynamicTableSinkFactoryTest {
createDynamicTableSink(options);
}
+ /**
+ * {@code retry.times} is the legacy DDL name of the send retry count. It
used to be accepted
+ * and then silently dropped, so a SQL job that configured it retried with
the default instead.
+ */
+ @Test
+ public void testLegacyRetryTimesReachesTheWriter() throws Exception {
+ Map<String, String> options = sinkOptions();
+ options.put(RocketMQSinkOptions.OPTIONAL_WRITE_RETRY_TIMES.key(), "7");
+
+
assertThat(sinkConfiguration(options).getInteger(RocketMQSinkOptions.SEND_RETRY_TIMES))
+ .isEqualTo(7);
+ }
+
+ @Test
+ public void testSendRetryTimesWinsOverTheLegacyAlias() throws Exception {
+ Map<String, String> options = sinkOptions();
+ options.put(RocketMQSinkOptions.OPTIONAL_WRITE_RETRY_TIMES.key(), "7");
+ options.put(RocketMQSinkOptions.SEND_RETRY_TIMES.key(), "5");
+
+
assertThat(sinkConfiguration(options).getInteger(RocketMQSinkOptions.SEND_RETRY_TIMES))
+ .isEqualTo(5);
+ }
+
+ @Test
+ public void testRetryDefaultsToThreeWhenNeitherKeyIsSet() throws Exception
{
+ assertThat(
+ sinkConfiguration(sinkOptions())
+
.getInteger(RocketMQSinkOptions.SEND_RETRY_TIMES))
+ .isEqualTo(3);
+ }
+
+ private static Map<String, String> sinkOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put("connector", IDENTIFIER);
+ options.put(RocketMQSinkOptions.TOPIC.key(), TOPIC);
+ options.put(RocketMQSinkOptions.PRODUCER_GROUP.key(), PRODUCER_GROUP);
+ options.put(RocketMQOptions.ENDPOINTS.key(), NAME_SERVER_ADDRESS);
+ return options;
+ }
+
+ private static Configuration sinkConfiguration(Map<String, String>
options) throws Exception {
+ DynamicTableSink tableSink = createDynamicTableSink(options);
+ SinkV2Provider provider =
+ (SinkV2Provider)
+ tableSink.getSinkRuntimeProvider(new
SinkRuntimeProviderContext(false));
+ RocketMQSink<?> sink = (RocketMQSink<?>) provider.createSink();
+ Field configuration =
RocketMQSink.class.getDeclaredField("configuration");
+ configuration.setAccessible(true);
+ return (Configuration) configuration.get(sink);
+ }
+
private static DynamicTableSink createDynamicTableSink(Map<String, String>
options) {
return FactoryUtil.createTableSink(
null,
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkMigrationTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkMigrationTest.java
index ad09869..2707f50 100644
---
a/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkMigrationTest.java
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/table/RocketMQDynamicTableSinkMigrationTest.java
@@ -53,7 +53,6 @@ class RocketMQDynamicTableSinkMigrationTest {
null, // dynamicColumn
String.valueOf((char) 1), // fieldDelimiter (default
SOH)
"UTF-8", // encoding
- 10, // retryTimes
5000L, // sleepTime
false, // isDynamicTag
true, // isDynamicTagIncluded
@@ -91,7 +90,6 @@ class RocketMQDynamicTableSinkMigrationTest {
null, // dynamicColumn
String.valueOf((char) 1), // fieldDelimiter
"UTF-8", // encoding
- 10, // retryTimes
5000L, // sleepTime
false, // isDynamicTag
true, // isDynamicTagIncluded
