This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/subscription-broker-self-healing in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit acccd0e34a4fecaabfd386480e2a3ea4846cbf56 Author: Caideyipi <[email protected]> AuthorDate: Mon Oct 26 21:39:52 2026 +0800 Repair missing subscription brokers from consumer metadata --- .../agent/SubscriptionConsumerAgent.java | 56 +++++- .../receiver/SubscriptionReceiverV1.java | 6 + ...ubscriptionConsumerAgentBrokerRecoveryTest.java | 222 +++++++++++++++++++++ .../receiver/SubscriptionReceiverV1Test.java | 33 +++ 4 files changed, 307 insertions(+), 10 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java index 8af488c5923..df48627715d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java @@ -116,11 +116,7 @@ public class SubscriptionConsumerAgent { // if consumer group meta does not exist on local agent if (Objects.isNull(metaInAgent)) { - SubscriptionAgent.broker().createPipeBrokerIfNotExist(consumerGroupId); - ConsensusSubscriptionSetupHandler.setupConsensusSubscriptions( - consumerGroupId, - metaFromCoordinator.getSubscribedTopicNames(), - metaFromCoordinator.visibleUnder(true)); + ensureConsumerGroupBrokerInternal(metaFromCoordinator); replaceConsumerGroupMeta(consumerGroupId, metaFromCoordinator); return; } @@ -145,15 +141,13 @@ public class SubscriptionConsumerAgent { } } - ConsensusSubscriptionSetupHandler.setupConsensusSubscriptions( - consumerGroupId, - metaFromCoordinator.getSubscribedTopicNames(), - metaFromCoordinator.visibleUnder(true)); + ensureConsumerGroupBrokerInternal(metaFromCoordinator); replaceConsumerGroupMeta(consumerGroupId, metaFromCoordinator); - // no need to create broker manually return; } + ensureConsumerGroupBrokerInternal(metaFromCoordinator); + // remove prefetching queues for topics unsubscribed by the consumer group final Set<String> topicsUnsubByGroup = ConsumerGroupMeta.getTopicsUnsubByGroup(metaInAgent, metaFromCoordinator); @@ -279,6 +273,48 @@ public class SubscriptionConsumerAgent { } } + /** + * Ensures that a non-empty consumer group has the local broker and its consensus subscriptions. + * + * <p>The coordinator metadata is authoritative here. In particular, an empty metadata must not + * recreate a broker or restore topics that were explicitly removed from the group. + */ + public void ensureConsumerGroupBroker(final String consumerGroupId) { + executeWithConsumerGroupLock( + consumerGroupId, + () -> { + final ConsumerGroupMeta consumerGroupMeta = getConsumerGroupMeta(consumerGroupId); + if (Objects.nonNull(consumerGroupMeta)) { + ensureConsumerGroupBrokerInternal(consumerGroupMeta); + } + }); + } + + private void ensureConsumerGroupBrokerInternal(final ConsumerGroupMeta consumerGroupMeta) { + if (consumerGroupMeta.isEmpty()) { + return; + } + + final String consumerGroupId = consumerGroupMeta.getConsumerGroupId(); + final boolean brokerExisted = SubscriptionAgent.broker().isBrokerExist(consumerGroupId); + SubscriptionAgent.broker().createPipeBrokerIfNotExist(consumerGroupId); + try { + // setupConsensusSubscriptions is idempotent and also repairs a missing consensus broker or + // queue when a pipe broker still exists. + ConsensusSubscriptionSetupHandler.setupConsensusSubscriptions( + consumerGroupId, + consumerGroupMeta.getSubscribedTopicNames(), + consumerGroupMeta.visibleUnder(true)); + } catch (final RuntimeException e) { + // Allow a later metadata push or handshake to retry after a partial setup failure. Preserve + // an existing broker because it may still contain usable queues for the current metadata. + if (!brokerExisted) { + SubscriptionAgent.broker().dropBroker(consumerGroupId); + } + throw e; + } + } + private ConsumerGroupMeta getConsumerGroupMeta(final String consumerGroupId) { acquireReadLock(); try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java index cd53d9376ab..93d309cc5db 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java @@ -454,6 +454,12 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver { consumerConfig); } + // A previous inactivity cleanup or metadata race may have removed the local broker while the + // coordinator still has a non-empty consumer-group metadata record. Reconcile that state before + // exposing the consumer to subsequent requests. The operation is idempotent and serialized with + // metadata pushes by SubscriptionConsumerAgent. + SubscriptionAgent.consumer().ensureConsumerGroupBroker(consumerGroupId); + activateConsumer(consumerConfig); final int dataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgentBrokerRecoveryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgentBrokerRecoveryTest.java new file mode 100644 index 00000000000..46dca73ed8b --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgentBrokerRecoveryTest.java @@ -0,0 +1,222 @@ +/* + * 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.iotdb.db.subscription.agent; + +import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue; +import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta; +import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta; +import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter; +import org.apache.iotdb.db.subscription.broker.ISubscriptionBroker; +import org.apache.iotdb.db.subscription.broker.SubscriptionBroker; +import org.apache.iotdb.db.subscription.task.subtask.SubscriptionSinkSubtask; +import org.apache.iotdb.pipe.api.event.Event; +import org.apache.iotdb.rpc.subscription.config.ConsumerConfig; +import org.apache.iotdb.rpc.subscription.config.ConsumerConstant; + +import org.junit.Test; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +public class SubscriptionConsumerAgentBrokerRecoveryTest { + + private static final long TEST_TIMEOUT_SECONDS = 5L; + + @Test + public void testNonEmptyMetadataRecreatesMissingBroker() { + final String consumerGroupId = unique("group"); + final String consumerId = unique("consumer"); + final SubscriptionConsumerAgent agent = SubscriptionAgent.consumer(); + try { + final ConsumerGroupMeta meta = createMeta(consumerGroupId, consumerId); + assertNull(agent.handleSingleConsumerGroupMetaChanges(meta)); + assertTrue(SubscriptionAgent.broker().isBrokerExist(consumerGroupId)); + + assertTrue(SubscriptionAgent.broker().dropBroker(consumerGroupId)); + assertFalse(SubscriptionAgent.broker().isBrokerExist(consumerGroupId)); + + agent.ensureConsumerGroupBroker(consumerGroupId); + + assertTrue(SubscriptionAgent.broker().isBrokerExist(consumerGroupId)); + assertTrue(agent.isConsumerExisted(consumerGroupId, consumerId)); + } finally { + agent.handleDropConsumerGroup(consumerGroupId); + } + } + + @Test + public void testEmptyMetadataDoesNotRecreateBroker() { + final String consumerGroupId = unique("group"); + final String consumerId = unique("consumer"); + final SubscriptionConsumerAgent agent = SubscriptionAgent.consumer(); + try { + final ConsumerGroupMeta meta = createMeta(consumerGroupId, consumerId); + assertNull(agent.handleSingleConsumerGroupMetaChanges(meta)); + assertTrue(SubscriptionAgent.broker().isBrokerExist(consumerGroupId)); + + final ConsumerGroupMeta emptyMeta = meta.deepCopy(); + emptyMeta.removeConsumer(consumerId); + assertTrue(emptyMeta.isEmpty()); + assertNull(agent.handleSingleConsumerGroupMetaChanges(emptyMeta)); + + assertFalse(SubscriptionAgent.broker().isBrokerExist(consumerGroupId)); + agent.ensureConsumerGroupBroker(consumerGroupId); + assertFalse(SubscriptionAgent.broker().isBrokerExist(consumerGroupId)); + } finally { + agent.handleDropConsumerGroup(consumerGroupId); + } + } + + @Test + public void testConcurrentMetadataPushAndBrokerEnsureCreateOneBroker() throws Exception { + final String consumerGroupId = unique("group"); + final String consumerId = unique("consumer"); + final SubscriptionConsumerAgent agent = SubscriptionAgent.consumer(); + final ExecutorService executor = Executors.newFixedThreadPool(2); + try { + final ConsumerGroupMeta meta = createMeta(consumerGroupId, consumerId); + assertNull(agent.handleSingleConsumerGroupMetaChanges(meta)); + assertTrue(SubscriptionAgent.broker().dropBroker(consumerGroupId)); + + final Future<?> metadataPush = + executor.submit(() -> agent.handleSingleConsumerGroupMetaChanges(meta.deepCopy())); + final Future<?> handshakeRepair = + executor.submit(() -> agent.ensureConsumerGroupBroker(consumerGroupId)); + metadataPush.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + handshakeRepair.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + + assertEquals(1, getBrokers(consumerGroupId).size()); + } finally { + executor.shutdownNow(); + executor.awaitTermination(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + agent.handleDropConsumerGroup(consumerGroupId); + } + } + + @Test + public void testBrokerAndExistingQueueSurviveProviderReplacementRepair() throws Exception { + final String consumerGroupId = unique("group"); + final String consumerId = unique("consumer"); + final String topicName = unique("topic"); + final SubscriptionConsumerAgent agent = SubscriptionAgent.consumer(); + try { + final ConsumerGroupMeta meta = createMeta(consumerGroupId, consumerId); + meta.addSubscription(consumerId, Collections.singleton(topicName)); + assertNull(agent.handleSingleConsumerGroupMetaChanges(meta)); + + bindPipeQueue(consumerGroupId, topicName); + final SubscriptionBroker brokerBeforeRepair = getPipeBroker(consumerGroupId); + assertTrue(brokerBeforeRepair.hasQueue(topicName)); + + // A provider replacement closes only its session. The server-side repair must preserve the + // existing broker and queue when it reconciles the same non-empty metadata. + agent.ensureConsumerGroupBroker(consumerGroupId); + + assertSame(brokerBeforeRepair, getPipeBroker(consumerGroupId)); + assertTrue(getPipeBroker(consumerGroupId).hasQueue(topicName)); + } finally { + agent.handleDropConsumerGroup(consumerGroupId); + } + } + + @Test + public void testCanceledTopicIsNotRestoredWhenBrokerIsRecreated() throws Exception { + final String consumerGroupId = unique("group"); + final String consumerId = unique("consumer"); + final String topicName = unique("topic"); + final SubscriptionConsumerAgent agent = SubscriptionAgent.consumer(); + try { + final ConsumerGroupMeta meta = createMeta(consumerGroupId, consumerId); + meta.addSubscription(consumerId, Collections.singleton(topicName)); + assertNull(agent.handleSingleConsumerGroupMetaChanges(meta)); + + bindPipeQueue(consumerGroupId, topicName); + assertTrue(getPipeBroker(consumerGroupId).hasQueue(topicName)); + + final ConsumerGroupMeta unsubscribedMeta = meta.deepCopy(); + unsubscribedMeta.removeSubscription(consumerId, Collections.singleton(topicName)); + assertNull(agent.handleSingleConsumerGroupMetaChanges(unsubscribedMeta)); + assertFalse(getPipeBroker(consumerGroupId).hasQueue(topicName)); + + assertTrue(SubscriptionAgent.broker().dropBroker(consumerGroupId)); + agent.ensureConsumerGroupBroker(consumerGroupId); + + assertFalse(getPipeBroker(consumerGroupId).hasQueue(topicName)); + assertFalse(agent.getAllSubscriptions().containsKey(consumerGroupId)); + } finally { + agent.handleDropConsumerGroup(consumerGroupId); + } + } + + private static ConsumerGroupMeta createMeta( + final String consumerGroupId, final String consumerId) { + final Map<String, String> attributes = new HashMap<>(); + attributes.put(ConsumerConstant.CONSUMER_ID_KEY, consumerId); + attributes.put(ConsumerConstant.CONSUMER_GROUP_ID_KEY, consumerGroupId); + final ConsumerConfig consumerConfig = new ConsumerConfig(attributes); + return new ConsumerGroupMeta( + consumerGroupId, 1L, new ConsumerMeta(consumerId, 1L, consumerConfig.getAttribute())); + } + + private static void bindPipeQueue(final String consumerGroupId, final String topicName) { + final UnboundedBlockingPendingQueue<Event> inputQueue = + new UnboundedBlockingPendingQueue<>(new PipeDataRegionEventCounter()); + final SubscriptionSinkSubtask subtask = + new SubscriptionSinkSubtask( + unique("task"), 1L, "", 0, inputQueue, null, topicName, consumerGroupId); + SubscriptionAgent.broker().bindPrefetchingQueue(subtask); + } + + @SuppressWarnings("unchecked") + private static List<ISubscriptionBroker> getBrokers(final String consumerGroupId) + throws Exception { + final Field field = SubscriptionBrokerAgent.class.getDeclaredField("consumerGroupIdToBrokers"); + field.setAccessible(true); + final Map<String, List<ISubscriptionBroker>> brokersByConsumerGroup = + (Map<String, List<ISubscriptionBroker>>) field.get(SubscriptionAgent.broker()); + return brokersByConsumerGroup.getOrDefault(consumerGroupId, Collections.emptyList()); + } + + private static SubscriptionBroker getPipeBroker(final String consumerGroupId) throws Exception { + return getBrokers(consumerGroupId).stream() + .filter(SubscriptionBroker.class::isInstance) + .map(SubscriptionBroker.class::cast) + .findFirst() + .orElseThrow(AssertionError::new); + } + + private static String unique(final String prefix) { + return prefix + '-' + UUID.randomUUID(); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java index 40482fe1b20..eead481b746 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java @@ -221,6 +221,39 @@ public class SubscriptionReceiverV1Test { Assert.assertTrue((boolean) getField(receiver, "consumerFenced")); } + @Test + public void testHandshakeRepairsMissingBrokerWithoutChangingConsumerMetadata() { + final String consumerGroupId = "group-" + UUID.randomUUID(); + final String consumerId = "consumer-" + UUID.randomUUID(); + final ConsumerConfig consumerConfig = + createConsumerConfig(1_000L, null, null, consumerId, consumerGroupId); + final ConsumerGroupMeta consumerGroupMeta = + new ConsumerGroupMeta( + consumerGroupId, + System.currentTimeMillis(), + new ConsumerMeta( + consumerId, System.currentTimeMillis(), consumerConfig.getAttribute())); + final SubscriptionReceiverV1 receiver = new SubscriptionReceiverV1(); + Assert.assertNull( + SubscriptionAgent.consumer().handleSingleConsumerGroupMetaChanges(consumerGroupMeta)); + try { + Assert.assertTrue(SubscriptionAgent.broker().dropBroker(consumerGroupId)); + Assert.assertFalse(SubscriptionAgent.broker().isBrokerExist(consumerGroupId)); + + final TSStatus status = + receiver + .handle(PipeSubscribeHandshakeReq.toTPipeSubscribeReq(consumerConfig)) + .getStatus(); + + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + Assert.assertTrue(SubscriptionAgent.broker().isBrokerExist(consumerGroupId)); + Assert.assertTrue( + SubscriptionAgent.consumer().isConsumerExisted(consumerGroupId, consumerId)); + } finally { + SubscriptionAgent.consumer().handleDropConsumerGroup(consumerGroupId); + } + } + @Test public void testNeverHandshakenConsumerStillReturnsMissingConsumerStatus() { final SubscriptionReceiverV1 receiver = new SubscriptionReceiverV1();
