somak2kai opened a new issue, #10630: URL: https://github.com/apache/rocketmq/issues/10630
### Before Creating the Bug Report - [x] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq/discussions). - [x] I have searched the [GitHub Issues](https://github.com/apache/rocketmq/issues) and [GitHub Discussions](https://github.com/apache/rocketmq/discussions) of this repository and believe that this is not a duplicate. - [x] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ. ### Runtime platform environment OS: Mac but i believe would be applicable to all. ### RocketMQ version branch : develop ### JDK Version _No response_ ### Describe the Bug A potential data race and TOCTOU inconsistency found in org.apache.rocketmq.client.impl.consumer.DefaultLitePullConsumerImpl DefaultLitePullConsumerImpl maintains a ConcurrentMap<MessageQueue, PullTaskImpl> taskTable that tracks which queues currently have a live pull loop running. Three methods mutate this map — updateAssignPullTask, updatePullTask, removePullTask — via a non-atomic "iterate/remove stale entries, then containsKey-check-and-put new entries" sequence (startPullTask), with no synchronization around any of it Example concrete failure modes, both silent (no exception, no log line pointing at the cause): - Duplicate pull tasks. The containsKey → put check in startPullTask is not atomic. Two threads can both observe containsKey(mq) == false for a newly assigned queue and each construct and schedule their own PullTaskImpl. Only one survives in taskTable; the other keeps running, untracked, feeding consumeRequestCache independently. - Orphaned queue. One thread's it.remove() can wipe out an entry the other thread just added, leaving a queue with no live pull task and no map entry. Symptom: that queue silently stops being polled — no error, no log noise — until the next rebalance happens to touch its assignment again (which may not happen if group membership stays stable). ### Steps to Reproduce You can execute the below test case in current develop branch using ``` mvn -pl client -am test -Dtest=DefaultLitePullConsumerImplTaskTableRaceTest \ -DfailIfNoTests=false -Dspotbugs.skip=true -Dcheckstyle.skip=true -Drat.skip=true ``` Test case file is here: ``` /* * 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.rocketmq.client.impl.consumer; import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer; import org.apache.rocketmq.common.message.MessageQueue; import org.junit.After; import org.junit.Before; import org.junit.Test; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.Collection; import java.util.Collections; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; import static org.junit.Assert.assertEquals; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; /** * Reproduces the taskTable race in DefaultLitePullConsumerImpl class. * * updateAssignPullTask, updatePullTask and startPullTask perform a non-atomic * "containsKey then put" on taskTable. This unit test file is intended to * surface a data race */ public class DefaultLitePullConsumerImplTaskTableRaceTest { private static final int REPETITIONS = 500; private DefaultLitePullConsumerImpl consumer; private ConcurrentMap<MessageQueue, Object> taskTable; private ExecutorService racers; private AtomicInteger scheduleCallsThisRound; private Method updateAssignPullTaskMethod; private Method updatePullTaskMethod; private Method removePullTaskMethod; @Before public void setUp() throws Exception { consumer = new DefaultLitePullConsumerImpl(new DefaultLitePullConsumer(), null); scheduleCallsThisRound = new AtomicInteger(0); racers = Executors.newFixedThreadPool(2); ScheduledThreadPoolExecutor mockExecutor = mock(ScheduledThreadPoolExecutor.class); doAnswer(invocation -> { scheduleCallsThisRound.incrementAndGet(); return null; }).when(mockExecutor).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class)); setPrivateField("scheduledThreadPoolExecutor", mockExecutor); // noinspection unchecked taskTable = (ConcurrentMap<MessageQueue, Object>) getPrivateField("taskTable"); updateAssignPullTaskMethod = resolveMethod("updateAssignPullTask", Collection.class); updatePullTaskMethod = resolveMethod("updatePullTask", String.class, Set.class); removePullTaskMethod = resolveMethod("removePullTask", String.class); } @After public void tearDown() { racers.shutdownNow(); } /** * Checks to see if concurrent * org.apache.rocketmq.client.impl.consumer.DefaultLitePullConsumerImpl.updatePullTask * method * does not double schedule on the same queue. * * @throws Exception */ @Test public void testUpdatePullTaskDoesNotDoubleScheduleSameQueue() throws Exception { String topic = "raceTopic"; MessageQueue mq = new MessageQueue(topic, "brokerA", 0); Set<MessageQueue> mqDivided = Collections.singleton(mq); Runnable action = () -> consumer.updateAssignQueueAndStartPullTask(topic, Collections.emptySet(), mqDivided); assertNoRaceCondition( null, action, action, () -> scheduleCallsThisRound.get() > 1, "Unsynchronized containsKey-then-put race in startPullTask() (via updatePullTask) " + "produced duplicate schedules."); } /** * Checks to see if concurrent * org.apache.rocketmq.client.impl.consumer.DefaultLitePullConsumerImpl.updateAssignPullTask * does not double schedule for the same queue. * * @throws Exception */ @Test public void testUpdateAssignPullTaskDoesNotDoubleScheduleSameQueue() throws Exception { String topic = "raceTopicAssign"; MessageQueue mq = new MessageQueue(topic, "brokerA", 0); Collection<MessageQueue> mqNewSet = Collections.singleton(mq); Runnable action = () -> invoke(updateAssignPullTaskMethod, mqNewSet); assertNoRaceCondition( null, action, action, () -> scheduleCallsThisRound.get() > 1, "Unsynchronized containsKey-then-put race in startPullTask() (via updateAssignPullTask, " + "the assign()/start() path) produced duplicate schedules."); } /** * Tests * org.apache.rocketmq.client.impl.consumer.DefaultLitePullConsumerImpl.removePullTask * is threadsafe. */ @Test public void testRemovePullTaskDoesNotCorruptUnrelatedTopicAddition() throws Exception { String removedTopic = "raceTopicRemove"; String keptTopic = "raceTopicKeep"; MessageQueue mqToRemove = new MessageQueue(removedTopic, "brokerA", 0); MessageQueue mqToKeep = new MessageQueue(keptTopic, "brokerA", 0); Runnable setup = () -> invoke(updateAssignPullTaskMethod, Collections.singleton(mqToRemove)); Runnable removeAction = () -> invoke(removePullTaskMethod, removedTopic); Runnable addAction = () -> invoke(updatePullTaskMethod, keptTopic, Collections.singleton(mqToKeep)); assertNoRaceCondition( setup, removeAction, addAction, () -> { boolean removedTopicStillPresent = taskTable.keySet().stream() .anyMatch(mq -> mq.getTopic().equals(removedTopic)); return removedTopicStillPresent || !taskTable.containsKey(mqToKeep) || scheduleCallsThisRound.get() != 1; }, "Shared taskTable lock corruption detected during concurrent add/remove operations."); } private void assertNoRaceCondition(Runnable roundSetup, Runnable action1, Runnable action2, Supplier<Boolean> failureCondition, String errorMessage) throws Exception { int failedRounds = 0; // in order to reproduce potential data race, run it REPETITIONS number of // times. for (int round = 0; round < REPETITIONS; round++) { taskTable.clear(); scheduleCallsThisRound.set(0); if (roundSetup != null) { roundSetup.run(); scheduleCallsThisRound.set(0); } raceTwo(action1, action2); if (failureCondition.get()) { failedRounds++; } } assertEquals(errorMessage + " Saw " + failedRounds + "/" + REPETITIONS + " corrupted rounds.", 0, failedRounds); } /** * Using a cyclicbarrier to gate tasks start at same time. * * @param first * @param second * @throws Exception */ private void raceTwo(Runnable first, Runnable second) throws Exception { CyclicBarrier barrier = new CyclicBarrier(2); CompletableFuture<Void> f1 = CompletableFuture.runAsync(() -> runAtBarrier(first, barrier), racers); CompletableFuture<Void> f2 = CompletableFuture.runAsync(() -> runAtBarrier(second, barrier), racers); CompletableFuture.allOf(f1, f2).get(5, TimeUnit.SECONDS); } private void runAtBarrier(Runnable action, CyclicBarrier barrier) { try { barrier.await(5, TimeUnit.SECONDS); } catch (Exception e) { throw new RuntimeException(e); } action.run(); } private void setPrivateField(String fieldName, Object value) throws Exception { Field field = DefaultLitePullConsumerImpl.class.getDeclaredField(fieldName); field.setAccessible(true); field.set(consumer, value); } private Object getPrivateField(String fieldName) throws Exception { Field field = DefaultLitePullConsumerImpl.class.getDeclaredField(fieldName); field.setAccessible(true); return field.get(consumer); } private Method resolveMethod(String name, Class<?>... parameterTypes) throws NoSuchMethodException { Method method = DefaultLitePullConsumerImpl.class.getDeclaredMethod(name, parameterTypes); method.setAccessible(true); return method; } private void invoke(Method method, Object... args) { try { method.invoke(consumer, args); } catch (Exception e) { throw new RuntimeException(e); } } } ``` ### What Did You Expect to See? Success test cases ### What Did You See Instead? Unit test fails with: ``` Tests run: 2, Failures: 1, Errors: 0, Skipped: 1, Time elapsed: 0.755 sec <<< FAILURE! - in org.apache.rocketmq.client.impl.consumer.DefaultLitePullConsumerImplTaskTableRaceTest testUpdateAssignPullTaskDoesNotDoubleScheduleSameQueue(org.apache.rocketmq.client.impl.consumer.DefaultLitePullConsumerImplTaskTableRaceTest) Time elapsed: 0.719 sec <<< FAILURE! java.lang.AssertionError: Unsynchronized containsKey-then-put race in startPullTask() (via updateAssignPullTask, the assign()/start() path) produced duplicate schedules. Saw 9/500 corrupted rounds. expected:<0> but was:<9> at org.apache.rocketmq.client.impl.consumer.DefaultLitePullConsumerImplTaskTableRaceTest.assertNoRaceCondition(DefaultLitePullConsumerImplTaskTableRaceTest.java:182) at org.apache.rocketmq.client.impl.consumer.DefaultLitePullConsumerImplTaskTableRaceTest.testUpdateAssignPullTaskDoesNotDoubleScheduleSameQueue(DefaultLitePullConsumerImplTaskTableRaceTest.java:130) Results : Failed tests: DefaultLitePullConsumerImplTaskTableRaceTest.testUpdateAssignPullTaskDoesNotDoubleScheduleSameQueue:130->assertNoRaceCondition:182 Unsynchronized containsKey-then-put race in startPullTask() (via updateAssignPullTask, the assign()/start() path) produced duplicate schedules. Saw 9/500 corrupted rounds. expected:<0> but was:<9> ``` ### Additional Context How i found this issue: In the sprit of full transparency : I found this while building my own open source case study [tool](https://github.com/somak2kai/beats) on java and golang structural fingerprints and potential outliers and I used rocketmq (among many others oss repos) as test subjects. The goal of the tool was not to find issues , but was to perform a case study. This particular issue stood out , hence raising it. No personal agenda behind this 🙏 These three methods were marked as outliers compared to their structural cousins where proper checks were performed. outliers marked : updateAssignPullTask | org.apache.rocketmq.client.impl.consumer | impl/consumer/DefaultLitePullConsumerImpl.java:465 -- | -- | -- removePullTask | org.apache.rocketmq.client.impl.consumer | impl/consumer/DefaultLitePullConsumerImpl.java:721 -- | -- | -- updatePullTask | org.apache.rocketmq.client.impl.consumer | impl/consumer/DefaultLitePullConsumerImpl.java:222 -- | -- | -- Compared to their structural cousins: updateAssignedMessageQueue | org.apache.rocketmq.client.impl.consumer | impl/consumer/AssignedMessageQueue.java -- | -- | -- updateAssignedMessageQueue | org.apache.rocketmq.client.impl.consumer | impl/consumer/AssignedMessageQueue.java -- | -- | -- removeAssignedMessageQueue | org.apache.rocketmq.client.impl.consumer | impl/consumer/AssignedMessageQueue.java -- | -- | -- -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
