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 80a98ac [ISSUE #151] fix(remoting): make split assignment and
partition discovery survive direct-connect route races (#152)
80a98ac is described below
commit 80a98acccef1dbefbb87bdd07f90bee6d90d91a1
Author: lizhimins <[email protected]>
AuthorDate: Thu Oct 29 13:59:16 2026 +0800
[ISSUE #151] fix(remoting): make split assignment and partition discovery
survive direct-connect route races (#152)
Two silent defects surface when the client talks to NameServers directly
(open-source deployment, no proxy):
1. Starting offsets are ignored for a random subset of queues.
RebalancePullImpl.computePullFromWhereWithException is hard-coded to 0,
so assign() lets the client-internal rebalance thread dispatch a first
pull from offset 0 while handleSplitsChanges is still seeking to the
configured starting offset. No Flink-side lock scope can win that race.
Enforce the starting offset deterministically at the fetch() outlet
instead: records below max(startingOffset, 0) are dropped there, and
seek() remains as a traffic optimization.
2. A NameServer outage restarts the job and rewinds offsets. The periodic
partition discovery flattened per-topic route fetch failures into an
empty queue set; diffed against the allocated set that revokes every
split, and the queues returning after recovery are re-initialized from
scratch (observed: +150 replayed records). A failed fetch says nothing
about the actual route: exclude failed topics from the round, skip the
round when every topic failed, and keep the current assignment when an
empty result arrives while queues are allocated. Genuine deletions
surface again on later rounds.
Also document that EXACTLY_ONCE is not supported on proxy-fronted
endpoints (transaction state lives in per-proxy-process memory and is
silently dropped), while direct deployments do support it.
Cluster verification of the outage scenario after the fix: job RUNNING
throughout the 60s NameServer stop, zero split decreases, zero replay
(sink grew exactly by the second wave), producer recovered within 150s
of the NameServer coming back.
---
docs/remoting-connector.md | 27 ++-
.../enumerator/RocketMQSourceEnumerator.java | 50 ++++--
.../source/reader/RocketMQSplitReader.java | 11 ++
...etMQSourceEnumeratorDiscoveryToleranceTest.java | 182 +++++++++++++++++++++
4 files changed, 247 insertions(+), 23 deletions(-)
diff --git a/docs/remoting-connector.md b/docs/remoting-connector.md
index c4c960a..5f555cd 100644
--- a/docs/remoting-connector.md
+++ b/docs/remoting-connector.md
@@ -50,13 +50,24 @@ 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.
+
+> **`EXACTLY_ONCE` is NOT supported when the endpoints point at a proxy**
(Alibaba Cloud
+> RocketMQ 5.x commercial instances, or any deployment where clients reach
brokers through a
+> proxy). Use `AT_LEAST_ONCE` with idempotent downstream consumers in those
deployments.
+>
+> Why: a proxy keeps the transaction state of a half message only in the
memory of the proxy
+> process that received it, and drops it after `transactionDataExpireMillis`
(30 seconds by
+> default). Flink commits at the end of a checkpoint, so a commit that lands
on a different
+> proxy pod, or arrives after the TTL, finds no state; the proxy then maps
+> `TRANSACTION_DATA_NOT_FOUND` to a **success** response and drops it without
a single log
+> line — the job sees successful commits while the records never become
visible, and nothing
+> surfaces anywhere. The broker-side transaction check does not rescue these
half messages
+> through the proxy path either. The failure is silent by construction, so it
cannot be
+> mitigated by tuning the checkpoint interval.
+>
+> `EXACTLY_ONCE` **is supported on direct deployments** (endpoints =
nameserver addresses, the
+> connector talks to brokers directly): the transaction state lives on the
broker that holds
+> the half message and is persisted, so commits succeed regardless of the
checkpoint interval.
## Client options (`rocketmq.client.*`)
@@ -128,7 +139,7 @@ group or a missing subscription relationship shows up.
| `rocketmq.sink.topic` | String | (none) | Topic, required for SQL |
| `rocketmq.sink.group` | String | `PID-flink-producer` | Producer group |
| `rocketmq.sink.tag` | String | (none) | Static message tag |
-| `rocketmq.sink.delivery.guarantee` | String | `AT_LEAST_ONCE` | `NONE` /
`AT_LEAST_ONCE` / `EXACTLY_ONCE` |
+| `rocketmq.sink.delivery.guarantee` | String | `AT_LEAST_ONCE` | `NONE` /
`AT_LEAST_ONCE` / `EXACTLY_ONCE` (`EXACTLY_ONCE` is **not supported via a proxy
endpoint**, see the delivery-semantics note above) |
| `rocketmq.sink.transaction.timeout` | Long | 900 (s) | Transaction timeout
for EXACTLY_ONCE |
| `rocketmq.sink.send.timeout` | Long | 5000 | Send timeout (ms) |
| `rocketmq.sink.send.retry.times` | Integer | 3 | Send retry times |
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumerator.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumerator.java
index 12e31f7..fb9841e 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumerator.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumerator.java
@@ -62,7 +62,6 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
-import java.util.stream.Stream;
/** The enumerator class for RocketMQ source. */
@Internal
@@ -281,20 +280,30 @@ public class RocketMQSourceEnumerator
.getString(RocketMQSourceOptions.TOPIC)
.split(RocketMQSourceOptions.TOPIC_SEPARATOR)));
- return topicSet.stream()
- .flatMap(
- topic -> {
- try {
- return
consumer.fetchMessageQueues(topic).get().stream();
- } catch (Exception e) {
- log.error(
- "Request topic route for service
discovery error, topic={}",
- topic,
- e);
- }
- return Stream.empty();
- })
- .collect(Collectors.toSet());
+ Set<MessageQueue> fetchedQueues = new HashSet<>();
+ Set<String> failedTopics = new HashSet<>();
+ for (String topic : topicSet) {
+ try {
+ fetchedQueues.addAll(consumer.fetchMessageQueues(topic).get());
+ } catch (Exception e) {
+ failedTopics.add(topic);
+ log.error("Request topic route for service discovery error,
topic={}", topic, e);
+ }
+ }
+ if (!failedTopics.isEmpty()) {
+ // A failed fetch says nothing about the current route: the topic
may be perfectly
+ // healthy behind an unreachable nameserver. Reporting its queues
as gone would
+ // revoke and later re-initialize their splits (mass rebalance
plus an offset
+ // rewind once the nameserver is back), so failed topics are
excluded from this
+ // round instead, and a round where every topic failed is skipped
entirely.
+ log.warn(
+ "Skipping partition decrease evaluation for topics whose
route fetch failed: {}",
+ failedTopics);
+ if (failedTopics.size() == topicSet.size()) {
+ return null;
+ }
+ }
+ return fetchedQueues;
}
// This method should only be invoked in the coordinator executor thread.
@@ -305,6 +314,17 @@ public class RocketMQSourceEnumerator
if (latestSet == null) {
return;
}
+ if (latestSet.isEmpty() && !allocatedSet.isEmpty()) {
+ // An empty discovery result while queues are allocated is
indistinguishable from
+ // "every route fetch failed" and almost never a real route
deletion; treating it
+ // as a decrease would revoke the whole assignment. Real deletions
surface again
+ // on later rounds and are handled together with the rest of the
change result.
+ log.warn(
+ "Partition discovery returned an empty queue set while {}
queues are allocated;"
+ + " keeping the current assignment for this
round.",
+ allocatedSet.size());
+ return;
+ }
final SourceChangeResult sourceChangeResult =
getSourceChangeResult(latestSet);
if (sourceChangeResult.isEmpty()) {
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 294b2d4..1ee1fb3 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
@@ -186,6 +186,17 @@ public class RocketMQSplitReader<T> implements
SplitReader<MessageView, RocketMQ
rocketmqSourceReaderMetrics.setCurrentOffset(
messageQueue, messageView.getQueueOffset() + 1);
rocketmqSourceReaderMetrics.recordBytesConsumed(messageView.getStoreSize());
+ // The LitePull client starts the first pull of a freshly
assigned queue at
+ // offset 0
(RebalancePullImpl.computePullFromWhereWithException is hard-coded
+ // to 0), and the seek() issued by handleSplitsChanges races
with that first
+ // pull on the client's rebalance/pull threads. Dropping
everything below the
+ // split's starting offset here is what makes the configured
start position
+ // deterministic regardless of who wins that race.
+ Tuple2<Long, Long> splitOffsets =
currentOffsetTable.get(messageQueue);
+ if (splitOffsets != null
+ && messageView.getQueueOffset() <
Math.max(splitOffsets.f0, 0L)) {
+ continue;
+ }
if (!matchesFilter(messageView)) {
continue;
}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumeratorDiscoveryToleranceTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumeratorDiscoveryToleranceTest.java
new file mode 100644
index 0000000..42a750b
--- /dev/null
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumeratorDiscoveryToleranceTest.java
@@ -0,0 +1,182 @@
+/*
+ * 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.enumerator;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.rocketmq.source.InnerConsumer;
+import org.apache.flink.connector.rocketmq.source.RocketMQSourceOptions;
+import
org.apache.flink.connector.rocketmq.source.enumerator.offset.OffsetsSelector;
+
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * A route fetch failure must never be interpreted as "those queues are gone":
during a nameserver
+ * outage the topic route is unchanged, and revoking the whole assignment
would cause a mass
+ * rebalance plus an offset rewind once the nameserver is back.
+ */
+public class RocketMQSourceEnumeratorDiscoveryToleranceTest {
+
+ private static final String BROKER = "broker-a";
+
+ @Test
+ public void allTopicsFetchFailedSkipsDiscoveryRoundTest() throws Exception
{
+ InnerConsumer consumer = mock(InnerConsumer.class);
+ when(consumer.fetchMessageQueues("t1"))
+ .thenReturn(failedFuture(new RuntimeException("nameserver
unreachable")));
+ when(consumer.fetchMessageQueues("t2"))
+ .thenReturn(failedFuture(new RuntimeException("nameserver
unreachable")));
+ RocketMQSourceEnumerator enumerator = newEnumerator(consumer, "t1;t2");
+
+ assertThat(invokeRequestServiceDiscovery(enumerator)).isNull();
+ }
+
+ @Test
+ public void partialFetchFailureKeepsFailedTopicQueuesTest() throws
Exception {
+ InnerConsumer consumer = mock(InnerConsumer.class);
+ when(consumer.fetchMessageQueues("t1"))
+ .thenReturn(
+ CompletableFuture.completedFuture(
+ Arrays.asList(queue("t1", 0), queue("t1",
1))));
+ when(consumer.fetchMessageQueues("t2"))
+ .thenReturn(failedFuture(new RuntimeException("nameserver
unreachable")));
+ RocketMQSourceEnumerator enumerator = newEnumerator(consumer, "t1;t2");
+
+ Set<MessageQueue> discovered =
invokeRequestServiceDiscovery(enumerator);
+
+ assertThat(discovered).containsExactlyInAnyOrder(queue("t1", 0),
queue("t1", 1));
+ }
+
+ @Test
+ public void successfulDiscoveryReturnsAllQueuesTest() throws Exception {
+ InnerConsumer consumer = mock(InnerConsumer.class);
+ when(consumer.fetchMessageQueues("t1"))
+ .thenReturn(
+ CompletableFuture.completedFuture(
+ Arrays.asList(queue("t1", 0), queue("t1",
1))));
+ when(consumer.fetchMessageQueues("t2"))
+
.thenReturn(CompletableFuture.completedFuture(Arrays.asList(queue("t2", 0))));
+ RocketMQSourceEnumerator enumerator = newEnumerator(consumer, "t1;t2");
+
+ Set<MessageQueue> discovered =
invokeRequestServiceDiscovery(enumerator);
+
+ assertThat(discovered)
+ .containsExactlyInAnyOrder(queue("t1", 0), queue("t1", 1),
queue("t2", 0));
+ }
+
+ @Test
+ public void emptyDiscoveryWithAllocatedQueuesKeepsAssignmentTest() throws
Exception {
+ InnerConsumer consumer = mock(InnerConsumer.class);
+ RocketMQSourceEnumerator enumerator = newEnumerator(consumer, "t1");
+ setAllocatedSet(enumerator, new HashSet<>(Arrays.asList(queue("t1",
0), queue("t1", 1))));
+
+ Method handle =
+ RocketMQSourceEnumerator.class.getDeclaredMethod(
+ "handleSourceQueueChange", Set.class, Throwable.class);
+ handle.setAccessible(true);
+ // Must return without touching the split assignment or scheduling any
re-initialization.
+ handle.invoke(enumerator, new HashSet<MessageQueue>(), null);
+
+ assertThat(allocatedSetOf(enumerator)).hasSize(2);
+ }
+
+ @Test
+ public void emptyLatestSetAloneWouldRevokeEverythingTest() throws
Exception {
+ // Documents the hazard the guards above prevent: diffing an empty
route against the
+ // allocated set reports every queue as decreased.
+ InnerConsumer consumer = mock(InnerConsumer.class);
+ RocketMQSourceEnumerator enumerator = newEnumerator(consumer, "t1");
+ setAllocatedSet(enumerator, new HashSet<>(Arrays.asList(queue("t1",
0), queue("t1", 1))));
+
+ Method diff =
+ RocketMQSourceEnumerator.class.getDeclaredMethod(
+ "getSourceChangeResult", Set.class);
+ diff.setAccessible(true);
+ Object changeResult = diff.invoke(enumerator, new
HashSet<MessageQueue>());
+
+ Field decreaseField =
changeResult.getClass().getDeclaredField("decreaseSet");
+ decreaseField.setAccessible(true);
+ assertThat((Collection<?>) decreaseField.get(changeResult)).hasSize(2);
+ }
+
+ private static RocketMQSourceEnumerator newEnumerator(InnerConsumer
consumer, String topics)
+ throws Exception {
+ Configuration configuration = new Configuration();
+ configuration.setString(RocketMQSourceOptions.TOPIC, topics);
+ RocketMQSourceEnumerator enumerator =
+ new RocketMQSourceEnumerator(
+ OffsetsSelector.earliest(),
+ OffsetsSelector.latest(),
+ Boundedness.CONTINUOUS_UNBOUNDED,
+ configuration,
+ null);
+ Field consumerField =
RocketMQSourceEnumerator.class.getDeclaredField("consumer");
+ consumerField.setAccessible(true);
+ consumerField.set(enumerator, consumer);
+ return enumerator;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Set<MessageQueue>
invokeRequestServiceDiscovery(RocketMQSourceEnumerator e)
+ throws Exception {
+ Method method =
RocketMQSourceEnumerator.class.getDeclaredMethod("requestServiceDiscovery");
+ method.setAccessible(true);
+ return (Set<MessageQueue>) method.invoke(e);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static void setAllocatedSet(RocketMQSourceEnumerator e,
Set<MessageQueue> queues)
+ throws Exception {
+ Field field =
RocketMQSourceEnumerator.class.getDeclaredField("allocatedSet");
+ field.setAccessible(true);
+ ((java.util.Map<MessageQueue, Byte>) field.get(e))
+ .putAll(queues.stream().collect(Collectors.toMap(q -> q, q ->
(byte) 0)));
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Set<MessageQueue> allocatedSetOf(RocketMQSourceEnumerator
e) throws Exception {
+ Field field =
RocketMQSourceEnumerator.class.getDeclaredField("allocatedSet");
+ field.setAccessible(true);
+ return ((java.util.Map<MessageQueue, Byte>) field.get(e)).keySet();
+ }
+
+ private static MessageQueue queue(String topic, int queueId) {
+ return new MessageQueue(topic, BROKER, queueId);
+ }
+
+ private static CompletableFuture<Collection<MessageQueue>>
failedFuture(Throwable t) {
+ CompletableFuture<Collection<MessageQueue>> future = new
CompletableFuture<>();
+ future.completeExceptionally(t);
+ return future;
+ }
+}
