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 1cbb6da [ISSUE #141] fix(legacy): stop the legacy source rewinding
and surface legacy sink async failures (#148)
1cbb6da is described below
commit 1cbb6daa046e275c2b43f6ff2a15ff3aebf5ccd5
Author: lizhimins <[email protected]>
AuthorDate: Mon Oct 26 20:38:37 2026 +0800
[ISSUE #141] fix(legacy): stop the legacy source rewinding and surface
legacy sink async failures (#148)
The legacy source read its next position back from
DefaultLitePullConsumer#committed
after every poll. Auto-commit is off by default, so the broker-side commit
never
advances and every poll round re-seeked to the same offset and re-emitted
the whole
topic: an acceptance run on a commercial 5.5.1 cluster turned 101 source
messages
into 120065 sink messages (118560 duplicates) while the job stayed RUNNING
with no
restarts and no errors. Book the position from the messages a poll drains
instead,
per queue, because poll serves every assigned queue.
Seeking, polling and booking now happen under one lock. poll drains all
assigned
queues, so a seek issued by one queue thread after another thread's poll
rewound
that queue and duplicated the messages it had just drained; the first
version of
this fix still showed 30 duplicates for that reason.
initOffsets also stored committed() verbatim, which is -1 for a group that
has
never committed. The EARLIEST and LATEST branches had no fallback for it
(only
GROUP_OFFSETS did), so the first seek failed with "Seek offset illegal" and
the
source gave up, finishing the job having read nothing. Fall back to the
position
the preceding seek left in the local offset store.
The legacy sink parked the first async send failure and rethrew it from
finish(),
and invoke() checks it too, but a burst of failures followed by an idle
stream
never reaches another invoke, and snapshotState did not check at all:
checkpoints
completed over the lost records and stop-with-savepoint succeeded. Check the
parked failure in snapshotState as well, so a checkpoint cannot commit a
loss.
Verified on a commercial 5.5.1 cluster with Flink 1.20.3: legacy source +
legacy
sink, 100 messages, duplicates 118560 -> 0; legacy sink in pure async mode
with
every send rejected, task failures 0 -> 3 with the IOException raised from
snapshotState, i.e. the loss is now loud instead of silent.
---
.../connectors/rocketmq/RocketMQSink.java | 5 ++
.../rocketmq/RocketMQSourceFunction.java | 90 ++++++++++++++++------
.../rocketmq/RocketMQSinkAsyncFailureTest.java | 90 ++++++++++++++++++++++
.../sourceFunction/RocketMQSourceFunctionTest.java | 66 ++++++++++++++++
4 files changed, 229 insertions(+), 22 deletions(-)
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSink.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSink.java
index f6cae6e..9f76290 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSink.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSink.java
@@ -258,6 +258,11 @@ public class RocketMQSink extends
RichSinkFunction<Message> implements Checkpoin
@Override
public void snapshotState(FunctionSnapshotContext context) throws
Exception {
flushSync();
+ // A checkpoint must not complete over records whose async send
already failed.
+ // invoke() only surfaces a parked failure when the next record
arrives, so a burst of
+ // failures followed by an idle stream would otherwise be checkpointed
as delivered:
+ // the job keeps running, every checkpoint succeeds, and the records
are gone.
+ checkAsyncSendException();
}
@Override
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSourceFunction.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSourceFunction.java
index 2fe7f98..56d8aac 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSourceFunction.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSourceFunction.java
@@ -49,6 +49,7 @@ import org.apache.commons.collections.map.LinkedMap;
import org.apache.commons.lang.Validate;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
+import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
@@ -296,31 +297,61 @@ public class RocketMQSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
private void pollAndEmit(MessageQueue mq, SourceContext<OUT> context, int
pullBatchSize)
throws Exception {
- Long offset = offsetTable.get(mq);
consumer.setPullBatchSize(pullBatchSize);
- consumer.seek(mq, offset);
-
- List<MessageExt> messages =
- consumer.poll(
- RocketMQUtils.getInteger(
- props,
- RocketMQConfig.CONSUMER_TIMEOUT,
- RocketMQConfig.DEFAULT_CONSUMER_TIMEOUT));
-
- boolean found = CollectionUtils.isNotEmpty(messages);
- if (found) {
- emitMessages(messages, context);
- }
-
+ List<MessageExt> messages;
+ // seek, poll and position booking have to be atomic across the
per-queue threads:
+ // poll drains every assigned queue, so a seek issued from one thread
after another
+ // thread's poll rewinds that queue and re-emits messages it already
drained.
synchronized (checkPointLock) {
- updateMessageQueueOffset(mq, consumer.committed(mq));
+ Long offset = offsetTable.get(mq);
+ if (offset != null && offset >= 0) {
+ consumer.seek(mq, offset);
+ }
+ messages =
+ consumer.poll(
+ RocketMQUtils.getInteger(
+ props,
+ RocketMQConfig.CONSUMER_TIMEOUT,
+ RocketMQConfig.DEFAULT_CONSUMER_TIMEOUT));
+ if (CollectionUtils.isNotEmpty(messages)) {
+ advanceConsumedOffsets(messages, mq);
+ }
}
- if (!found) {
+ if (CollectionUtils.isNotEmpty(messages)) {
+ emitMessages(messages, context);
+ } else {
RetryUtil.waitForMs(RocketMQConfig.DEFAULT_CONSUMER_DELAY_WHEN_MESSAGE_NOT_FOUND);
}
}
+ /**
+ * Books the next offset to read for every queue this poll drained. {@code
poll} serves all
+ * assigned queues, so each message is booked under its own queue rather
than under the queue
+ * this thread was handed.
+ *
+ * <p>The position must come from the messages themselves. Reading it back
from {@code
+ * DefaultLitePullConsumer#committed} rewinds the source whenever the
broker-side commit does
+ * not advance — auto-commit is off by default — which makes every poll
round re-seek to the
+ * same position and re-emit the whole topic, multiplying the output by
the number of rounds
+ * while every health signal stays green.
+ */
+ private void advanceConsumedOffsets(List<MessageExt> messages,
MessageQueue assigned)
+ throws MQClientException {
+ for (MessageExt message : messages) {
+ MessageQueue queue =
+ new MessageQueue(
+ assigned.getTopic(), message.getBrokerName(),
message.getQueueId());
+ long nextOffset = message.getQueueOffset() + 1;
+ Long current = offsetTable.get(queue);
+ // Only move forward: a seek in flight on another thread can let a
poll hand
+ // back a message some thread already booked.
+ if (current == null || nextOffset > current) {
+ updateMessageQueueOffset(queue, nextOffset);
+ }
+ }
+ }
+
private void emitMessages(List<MessageExt> messages, SourceContext<OUT>
context) {
long fetchTime = System.currentTimeMillis();
for (MessageExt msg : messages) {
@@ -360,17 +391,32 @@ public class RocketMQSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
* @param messageQueues
* @throws MQClientException
*/
+ /**
+ * The position a preceding {@code seekToBegin}/{@code seekToEnd}
established. {@code committed}
+ * answers -1 for a group that has never committed, and storing that -1
makes the first {@code
+ * seek} in the poll loop fail with "Seek offset illegal", after which the
source gives up on
+ * the queue and the job finishes having read nothing.
+ */
+ private long positionAfterSeek(MessageQueue mq) throws MQClientException {
+ long offset =
+ consumer.getOffsetStore().readOffset(mq,
ReadOffsetType.MEMORY_FIRST_THEN_STORE);
+ if (offset < 0) {
+ offset = Math.max(consumer.committed(mq), 0L);
+ }
+ return offset;
+ }
+
private void initOffsets(List<MessageQueue> messageQueues) throws
MQClientException {
for (MessageQueue mq : messageQueues) {
long offset;
switch (startMode) {
case LATEST:
consumer.seekToEnd(mq);
- offset = consumer.committed(mq);
+ offset = positionAfterSeek(mq);
break;
case EARLIEST:
consumer.seekToBegin(mq);
- offset = consumer.committed(mq);
+ offset = positionAfterSeek(mq);
break;
case GROUP_OFFSETS:
offset = consumer.committed(mq);
@@ -382,7 +428,7 @@ public class RocketMQSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
switch (offsetResetStrategy) {
case LATEST:
consumer.seekToEnd(mq);
- offset = consumer.committed(mq);
+ offset = positionAfterSeek(mq);
log.info(
"current consumer thread:{} has no
committed offset,use Strategy:{} instead",
mq,
@@ -394,7 +440,7 @@ public class RocketMQSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
mq,
offsetResetStrategy);
consumer.seekToBegin(mq);
- offset = consumer.committed(mq);
+ offset = positionAfterSeek(mq);
break;
default:
break;
@@ -413,7 +459,7 @@ public class RocketMQSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
if (specificOffset != null) {
offset = specificOffset;
} else {
- offset = consumer.committed(mq);
+ offset = Math.max(consumer.committed(mq), 0L);
}
break;
default:
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSinkAsyncFailureTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSinkAsyncFailureTest.java
new file mode 100644
index 0000000..8dab4cf
--- /dev/null
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSinkAsyncFailureTest.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.streaming.connectors.rocketmq;
+
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.connectors.rocketmq.common.util.TestUtils;
+
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.common.message.Message;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.Properties;
+
+import static org.junit.Assert.fail;
+
+/**
+ * The legacy sink parks the first asynchronous send failure and rethrows it
later. These tests pin
+ * down <em>where</em> it is rethrown: a checkpoint must not complete over
records whose send
+ * already failed, otherwise an idle or finished stream commits the loss while
every health signal
+ * stays green.
+ */
+public class RocketMQSinkAsyncFailureTest {
+
+ private static final FunctionSnapshotContext SNAPSHOT =
+ new FunctionSnapshotContext() {
+ @Override
+ public long getCheckpointId() {
+ return 1L;
+ }
+
+ @Override
+ public long getCheckpointTimestamp() {
+ return 1L;
+ }
+ };
+
+ @Test
+ public void snapshotStateSurfacesParkedAsyncSendFailureTest() throws
Exception {
+ RocketMQSink sink = new RocketMQSink(new Properties());
+ DefaultMQProducer producer = Mockito.mock(DefaultMQProducer.class);
+ Mockito.doAnswer(
+ invocation -> {
+ SendCallback callback = invocation.getArgument(1);
+ callback.onException(new IOException("CODE: 13
message rejected"));
+ return null;
+ })
+ .when(producer)
+ .send(Mockito.any(Message.class),
Mockito.any(SendCallback.class));
+ TestUtils.setFieldValue(sink, "producer", producer);
+ TestUtils.setFieldValue(sink, "async", true);
+ TestUtils.setFieldValue(
+ sink, "sinkInTps",
Mockito.mock(org.apache.flink.metrics.Meter.class));
+
+ // The burst of records is sent before any callback fires, so invoke()
itself
+ // returns normally and only parks the failure.
+ sink.invoke(new Message("topic", "body".getBytes()), null);
+
+ try {
+ sink.snapshotState(SNAPSHOT);
+ fail("snapshotState must fail the checkpoint when an async send
already failed");
+ } catch (IOException expected) {
+ // the parked failure must surface here, not be committed over
+ }
+ }
+
+ @Test
+ public void snapshotStatePassesWhenNoSendFailedTest() throws Exception {
+ RocketMQSink sink = new RocketMQSink(new Properties());
+ sink.snapshotState(SNAPSHOT);
+ }
+}
diff --git
a/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/sourceFunction/RocketMQSourceFunctionTest.java
b/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/sourceFunction/RocketMQSourceFunctionTest.java
index af70aac..5b605cc 100644
---
a/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/sourceFunction/RocketMQSourceFunctionTest.java
+++
b/flink-connector-rocketmq/src/test/java/org/apache/flink/streaming/connectors/rocketmq/sourceFunction/RocketMQSourceFunctionTest.java
@@ -18,6 +18,8 @@
package org.apache.flink.streaming.connectors.rocketmq.sourceFunction;
+import org.apache.flink.metrics.Meter;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.rocketmq.RocketMQConfig;
import org.apache.flink.streaming.connectors.rocketmq.RocketMQSourceFunction;
import
org.apache.flink.streaming.connectors.rocketmq.common.config.OffsetResetStrategy;
@@ -26,12 +28,17 @@ import
org.apache.flink.streaming.connectors.rocketmq.common.serialization.Simpl
import org.apache.flink.streaming.connectors.rocketmq.common.util.TestUtils;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
+import org.apache.rocketmq.client.consumer.store.OffsetStore;
+import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -74,6 +81,9 @@ public class RocketMQSourceFunctionTest {
public void testRestartFromCheckpoint() throws Exception {
DefaultLitePullConsumer consumer =
Mockito.mock(DefaultLitePullConsumer.class);
Mockito.when(consumer.committed(Mockito.any())).thenReturn(40L);
+ OffsetStore offsetStore = Mockito.mock(OffsetStore.class);
+ Mockito.when(offsetStore.readOffset(Mockito.any(),
Mockito.any())).thenReturn(40L);
+ Mockito.when(consumer.getOffsetStore()).thenReturn(offsetStore);
Properties properties = new Properties();
properties.setProperty(RocketMQConfig.CONSUMER_GROUP,
"${ConsumerGroup}");
properties.setProperty(RocketMQConfig.CONSUMER_TOPIC,
"${SourceTopic}");
@@ -102,4 +112,60 @@ public class RocketMQSourceFunctionTest {
}
assertEquals(offsetTable.containsKey(newMessageQueue), true);
}
+
+ /**
+ * The read position must advance from the messages a poll drained, never
from {@code
+ * committed(mq)}: with auto-commit off the broker-side commit does not
advance, and reading the
+ * position back from it makes every poll round re-seek to the same offset
and re-emit the whole
+ * topic (an observed 1189x duplication with a perfectly healthy,
never-restarting job).
+ */
+ @Test
+ public void testPollAdvancesOffsetsFromConsumedMessages() throws Exception
{
+ MessageQueue mq = new MessageQueue("tpc", "broker-0", 0);
+ DefaultLitePullConsumer consumer =
Mockito.mock(DefaultLitePullConsumer.class);
+ // A stale broker-side commit that never advances.
+ Mockito.when(consumer.committed(Mockito.any())).thenReturn(0L);
+ Mockito.when(consumer.poll(Mockito.anyLong()))
+ .thenReturn(Collections.singletonList(message(mq, 5L)))
+ .thenReturn(Collections.emptyList());
+
+ RocketMQSourceFunction<String> source =
+ new RocketMQSourceFunction<>(
+ new SimpleStringDeserializationSchema(), new
Properties());
+ ConcurrentHashMap<MessageQueue, Long> offsetTable = new
ConcurrentHashMap<>();
+ offsetTable.put(mq, 5L);
+ TestUtils.setFieldValue(source, "consumer", consumer);
+ TestUtils.setFieldValue(source, "checkPointLock", new Object());
+ TestUtils.setFieldValue(source, "enableCheckpoint", true);
+ TestUtils.setFieldValue(source, "offsetTable", offsetTable);
+ TestUtils.setFieldValue(source, "tpsMetric",
Mockito.mock(Meter.class));
+
+ Method pollAndEmit =
+ RocketMQSourceFunction.class.getDeclaredMethod(
+ "pollAndEmit",
+ MessageQueue.class,
+ SourceFunction.SourceContext.class,
+ int.class);
+ pollAndEmit.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ SourceFunction.SourceContext<String> context =
+ Mockito.mock(SourceFunction.SourceContext.class);
+
+ pollAndEmit.invoke(source, mq, context, 32);
+ assertEquals(Long.valueOf(6L), offsetTable.get(mq));
+
+ // An empty poll must leave the position alone instead of rewinding to
committed().
+ pollAndEmit.invoke(source, mq, context, 32);
+ assertEquals(Long.valueOf(6L), offsetTable.get(mq));
+ }
+
+ private static MessageExt message(MessageQueue mq, long queueOffset) {
+ MessageExt message = new MessageExt();
+ message.setTopic(mq.getTopic());
+ message.setBrokerName(mq.getBrokerName());
+ message.setQueueId(mq.getQueueId());
+ message.setQueueOffset(queueOffset);
+ message.setBody("payload".getBytes(StandardCharsets.UTF_8));
+ return message;
+ }
}
