From ef886dc65d2bc893b7a6ef3d2f7094cbfad3cb35 Mon Sep 17 00:00:00 2001 From: 761417898 <761417898@qq.com> Date: Fri, 13 Nov 2026 20:52:56 +0800 Subject: [PATCH] [IoTConsensus] Preserve WAL needed by reconstructed peers --- .../commit/IoTDBRegionReconstructWALIT.java | 216 ++++++++++++++++++ .../consensus/i18n/IoTConsensusMessages.java | 11 + .../consensus/i18n/IoTConsensusMessages.java | 11 + .../iotdb/consensus/iot/IoTConsensus.java | 1 + .../consensus/iot/IoTConsensusServerImpl.java | 86 ++++--- .../iot/logdispatcher/LogDispatcher.java | 91 ++++++-- .../iot/IoTConsensusServerImplTest.java | 95 ++++++++ .../dataregion/wal/node/WALNode.java | 60 +++-- .../dataregion/wal/node/WALNodeTest.java | 64 ++++++ 9 files changed, 561 insertions(+), 74 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionReconstructWALIT.java diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionReconstructWALIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionReconstructWALIT.java new file mode 100644 index 0000000000000..a1d364825ff64 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionReconstructWALIT.java @@ -0,0 +1,216 @@ +/* + * 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.confignode.it.regionmigration.pass.commit; + +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.it.utils.NodeDebugger; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.env.cluster.config.MppCommonConfig; +import org.apache.iotdb.it.env.cluster.config.MppDataNodeConfig; +import org.apache.iotdb.it.env.cluster.config.MppJVMConfig; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.ClusterIT; + +import com.sun.jdi.ObjectReference; +import com.sun.jdi.StackFrame; +import com.sun.jdi.ThreadReference; +import org.apache.tsfile.utils.Pair; +import org.awaitility.Awaitility; +import org.junit.After; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.net.ServerSocket; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework.getDataRegionMapWithLeader; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +/** Exercises WAL reclamation while the replacement replica cannot yet accept incrementals. */ +@RunWith(IoTDBTestRunner.class) +@Category(ClusterIT.class) +public class IoTDBRegionReconstructWALIT { + private static final String SERVER = "org.apache.iotdb.consensus.iot.IoTConsensusServerImpl"; + private static final int ROWS = 12000; + + @After + public void tearDown() { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testReconstructRetainsIncrementalsDuringSnapshot() throws Exception { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS) + .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) + .setDataReplicationFactor(2) + .setSchemaReplicationFactor(3) + .setDefaultDataRegionGroupNumPerDatabase(1) + .setDataRegionGroupExtensionPolicy("CUSTOM"); + Properties properties = new Properties(); + properties.setProperty("wal_file_size_threshold_in_byte", "16384"); + // Cleanup is invoked explicitly at the controlled snapshot boundary below. + properties.setProperty("delete_wal_files_period_in_ms", "3600000"); + ((MppCommonConfig) EnvFactory.getEnv().getConfig().getDataNodeCommonConfig()) + .updateProperties(properties); + EnvFactory.getEnv().initClusterEnvironment(1, 3); + Map ports = new HashMap<>(); + for (DataNodeWrapper node : EnvFactory.getEnv().getDataNodeWrapperList()) { + int port; + try (ServerSocket socket = new ServerSocket(0)) { + port = socket.getLocalPort(); + } + node.stopForcibly(); + MppJVMConfig jvm = new MppJVMConfig(); + jvm.setDebugPort(port); + node.changeConfig( + new MppDataNodeConfig(), + (MppCommonConfig) EnvFactory.getEnv().getConfig().getDataNodeCommonConfig(), + jvm); + node.start(); + ports.put(node, port); + } + EnvFactory.getAbstractEnv().checkClusterStatusWithoutUnknown(); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + final Map>> regions; + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("create timeseries root.rebuild.d.s with datatype=INT64, encoding=PLAIN"); + statement.execute("insert into root.rebuild.d(time,s) values(0,0)"); + statement.execute("flush"); + regions = getDataRegionMapWithLeader(statement); + } + assertEquals(1, regions.size()); + int region = regions.keySet().iterator().next(); + Pair> peers = regions.get(region); + int sourceId = peers.left; + int targetId = peers.right.stream().filter(id -> id != sourceId).findFirst().get(); + DataNodeWrapper source = EnvFactory.getEnv().dataNodeIdToWrapper(sourceId).get(); + DataNodeWrapper target = EnvFactory.getEnv().dataNodeIdToWrapper(targetId).get(); + DataNodeWrapper coordinator = + EnvFactory.getEnv().getDataNodeWrapperList().stream() + .filter(node -> node != source && node != target) + .findFirst() + .get(); + try (Connection stableConnection = EnvFactory.getEnv().getConnection(coordinator); + Statement stableStatement = stableConnection.createStatement(); + NodeDebugger debugger = new NodeDebugger(ports.get(source), SERVER, "transmitSnapshot")) { + Future reconstruct = + executor.submit( + () -> { + try (Connection c = EnvFactory.getEnv().getConnection(coordinator); + Statement s = c.createStatement()) { + s.execute("reconstruct region " + region + " on " + targetId); + } + return null; + }); + // Snapshot creation is complete; only its transmission thread is paused. The replacement + // remains inactive while ordinary writes and the real WAL cleaner keep running. + StackFrame frame = debugger.await(candidate -> true); + ThreadReference thread = frame.thread(); + ObjectReference server = frame.thisObject(); + ObjectReference wal = + (ObjectReference) + server.getValue(server.referenceType().fieldByName("consensusReqReader")); + for (int row = 1; row <= ROWS; row++) { + stableStatement.execute( + "insert into root.rebuild.d(time,s) values(" + row + "," + row + ")"); + } + // More requests than the dispatcher pipeline/queue can retain: replay must read WAL. + stableStatement.execute("flush"); + wal.invokeMethod( + thread, + wal.referenceType().methodsByName("rollWALFile").get(0), + Collections.emptyList(), + ObjectReference.INVOKE_SINGLE_THREADED); + wal.invokeMethod( + thread, + wal.referenceType().methodsByName("deleteOutdatedFiles").get(0), + Collections.emptyList(), + ObjectReference.INVOKE_SINGLE_THREADED); + debugger.resume(); + reconstruct.get(3, TimeUnit.MINUTES); + Awaitility.await() + .atMost(3, TimeUnit.MINUTES) + .until( + () -> { + try (ResultSet status = stableStatement.executeQuery("show regions")) { + while (status.next()) { + if (status.getInt("RegionId") == region + && status.getInt("DataNodeId") == targetId) { + return "Running".equals(status.getString("Status")); + } + } + return false; + } + }); + // Isolate each data replica. Connecting to different coordinators alone does not guarantee + // that the queries read different replicas. + source.stopForcibly(); + assertRows(stableStatement); + source.start(); + EnvFactory.getAbstractEnv().checkNodeInStatus(sourceId, NodeStatus.Running); + target.stopForcibly(); + assertRows(stableStatement); + } + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + } + } + + private static void assertRows(Statement statement) { + Awaitility.await() + .ignoreExceptionsInstanceOf(SQLException.class) + .atMost(1, TimeUnit.MINUTES) + .untilAsserted( + () -> { + try (ResultSet rows = + statement.executeQuery("select s from root.rebuild.d order by time")) { + int expected = 0; + while (rows.next()) { + assertEquals(expected, rows.getLong(1)); + assertEquals(expected, rows.getLong(2)); + expected++; + } + assertEquals(ROWS + 1, expected); + } + }); + } +} diff --git a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java index d07373015a83e..8aac42ee593a1 100644 --- a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java +++ b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java @@ -354,4 +354,15 @@ private IoTConsensusMessages() {} public static final String EXCEPTION_FAILED_TO_LOAD_SNAPSHOT_ARG_FOR_CONSENSUS_GROUP_ARG_F876AE5C = "Failed to load snapshot %s for consensus group %s"; + + public static final String EXCEPTION_NO_REPLICATION_CHANNEL_TO_ARG_F99CF90B = + "No replication channel to %s"; + public static final String EXCEPTION_REPLICATION_CHANNEL_TO_ARG_FAILED_402B2EC5 = + "Replication channel to %s failed"; + public static final String EXCEPTION_TIMED_OUT_WAITING_FOR_ARG_TO_APPLY_INDEX_ARG_ECAC0A5A = + "Timed out waiting for %s to apply index %d"; + public static final String EXCEPTION_INTERRUPTED_WAITING_FOR_REPLICATION_TO_ARG_64450D09 = + "Interrupted waiting for replication to %s"; + public static final String EXCEPTION_MISSING_WAL_ENTRIES_FOR_ARG_EXPECTED_ARG_FOUND_ARG_A9D194A5 = + "Missing WAL entries for %s: expected %d, found %d"; } diff --git a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java index 0d908aa8613de..ebd96ada7f8e0 100644 --- a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java +++ b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java @@ -349,4 +349,15 @@ private IoTConsensusMessages() {} public static final String EXCEPTION_FAILED_TO_LOAD_SNAPSHOT_ARG_FOR_CONSENSUS_GROUP_ARG_F876AE5C = "加载快照 %s 失败，共识组：%s"; + + public static final String EXCEPTION_NO_REPLICATION_CHANNEL_TO_ARG_F99CF90B = + "不存在到 %s 的复制通道"; + public static final String EXCEPTION_REPLICATION_CHANNEL_TO_ARG_FAILED_402B2EC5 = + "到 %s 的复制通道失败"; + public static final String EXCEPTION_TIMED_OUT_WAITING_FOR_ARG_TO_APPLY_INDEX_ARG_ECAC0A5A = + "等待 %s 应用索引 %d 超时"; + public static final String EXCEPTION_INTERRUPTED_WAITING_FOR_REPLICATION_TO_ARG_64450D09 = + "等待向 %s 复制时被中断"; + public static final String EXCEPTION_MISSING_WAL_ENTRIES_FOR_ARG_EXPECTED_ARG_FOUND_ARG_A9D194A5 = + "向 %s 复制所需的 WAL 条目缺失：预期索引 %d，实际索引 %d"; } diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java index 28b0458c8b6f4..a52f240514725 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java @@ -471,6 +471,7 @@ public void addRemotePeer(ConsensusGroupId groupId, Peer peer) throws ConsensusE // step 6: active new Peer logger.info(IoTConsensusMessages.ACTIVATE_NEW_PEER); impl.activePeer(peer); + impl.waitNewPeerUntilSyncLogCompleted(peer); // step 7: notify remote peer to clean up transferred snapshot logger.info(IoTConsensusMessages.CLEANUP_REMOTE_SNAPSHOT); diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java index df7126cde4f1d..dda7c2b022944 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java @@ -151,6 +151,7 @@ public class IoTConsensusServerImpl { private final Set configuration = ConcurrentHashMap.newKeySet(); private final AtomicLong searchIndex; + private final Object retentionPublicationLock = new Object(); private final LogDispatcher logDispatcher; private IoTConsensusConfig config; private final ConsensusReqReader consensusReqReader; @@ -902,6 +903,10 @@ public void waitTargetPeerUntilSyncLogCompleted(Peer targetPeer) } } + public void waitNewPeerUntilSyncLogCompleted(Peer peer) throws ConsensusGroupModifyPeerException { + logDispatcher.waitUntilSyncLogCompleted(peer, getSearchIndex(), TimeUnit.MINUTES.toMillis(10)); + } + public boolean hasReleaseAllRegionRelatedResource(ConsensusGroupId groupId) { return stateMachine.hasReleaseAllRegionRelatedResource(groupId); } @@ -975,23 +980,34 @@ private TSStatus syncWriterSafeTimeBarrierToPeer( /** build SyncLog channel with safeIndex as the default initial sync index. */ public void buildSyncLogChannel(Peer targetPeer, boolean startNow) { - buildSyncLogChannel(targetPeer, getMinSyncIndex(), startNow); + // Do not allow writes beyond the initial index until WAL retention protects the new peer. + stateMachineLock.lock(); + try { + synchronized (retentionPublicationLock) { + buildSyncLogChannel(targetPeer, getMinSyncIndex(), startNow); + } + } finally { + stateMachineLock.unlock(); + } } public void buildSyncLogChannel(Peer targetPeer, long initialSyncIndex, boolean startNow) { - KillPoint.setKillPoint(DataNodeKillPoints.ORIGINAL_ADD_PEER_DONE); - configuration.add(targetPeer); - if (Objects.equals(targetPeer, thisNode)) { - return; + synchronized (retentionPublicationLock) { + KillPoint.setKillPoint(DataNodeKillPoints.ORIGINAL_ADD_PEER_DONE); + configuration.add(targetPeer); + if (Objects.equals(targetPeer, thisNode)) { + return; + } + logDispatcher.addLogDispatcherThread(targetPeer, initialSyncIndex, startNow); + checkAndUpdateSafeDeletedSearchIndex(); + logger.info( + IoTConsensusMessages.BUILD_SYNC_LOG_CHANNEL_SUCCESS, + targetPeer, + initialSyncIndex, + startNow + ? IoTConsensusMessages.SYNC_LOG_CHANNEL_STARTED + : IoTConsensusMessages.SYNC_LOG_CHANNEL_START_LATER); } - logDispatcher.addLogDispatcherThread(targetPeer, initialSyncIndex, startNow); - logger.info( - IoTConsensusMessages.BUILD_SYNC_LOG_CHANNEL_SUCCESS, - targetPeer, - initialSyncIndex, - startNow - ? IoTConsensusMessages.SYNC_LOG_CHANNEL_STARTED - : IoTConsensusMessages.SYNC_LOG_CHANNEL_START_LATER); } /** @@ -1469,32 +1485,34 @@ void checkAndUpdateIndex() { * configured thresholds. */ public void checkAndUpdateSafeDeletedSearchIndex() { - if (configuration.isEmpty()) { - logger.error(IoTConsensusMessages.CONFIGURATION_EMPTY_UNEXPECTED); - return; - } + synchronized (retentionPublicationLock) { + if (configuration.isEmpty()) { + logger.error(IoTConsensusMessages.CONFIGURATION_EMPTY_UNEXPECTED); + return; + } - final boolean hasSubscriptions = !subscriptionQueueRegistry.isEmpty(); + final boolean hasSubscriptions = !subscriptionQueueRegistry.isEmpty(); - if (configuration.size() == 1 && !hasSubscriptions) { - // Single replica, no subscription consumers => delete all WAL freely - consensusReqReader.setSafelyDeletedSearchIndex(Long.MAX_VALUE); - consensusReqReader.setSubscriptionRetainedMinVersionId(Long.MAX_VALUE); - return; - } + if (configuration.size() == 1 && !hasSubscriptions) { + // Single replica, no subscription consumers => delete all WAL freely + consensusReqReader.setSafelyDeletedSearchIndex(Long.MAX_VALUE); + consensusReqReader.setSubscriptionRetainedMinVersionId(Long.MAX_VALUE); + return; + } - final long replicationIndex = - configuration.size() > 1 ? getMinFlushedSyncIndex() : Long.MAX_VALUE; + final long replicationIndex = + configuration.size() > 1 ? getMinFlushedSyncIndex() : Long.MAX_VALUE; - final SubscriptionRetentionBound subscriptionRetentionBound = - subscriptionWalRetentionCalculator.calculate( - subscriptionQueueRegistry.getRetentionPolicies(), - subscriptionQueueRegistry.getRetainedMinVersionIds()); + final SubscriptionRetentionBound subscriptionRetentionBound = + subscriptionWalRetentionCalculator.calculate( + subscriptionQueueRegistry.getRetentionPolicies(), + subscriptionQueueRegistry.getRetainedMinVersionIds()); - consensusReqReader.setSafelyDeletedSearchIndex( - Math.min(replicationIndex, subscriptionRetentionBound.getSafelyDeletedSearchIndex())); - consensusReqReader.setSubscriptionRetainedMinVersionId( - subscriptionRetentionBound.getRetainedMinVersionId()); + consensusReqReader.setSafelyDeletedSearchIndex( + Math.min(replicationIndex, subscriptionRetentionBound.getSafelyDeletedSearchIndex())); + consensusReqReader.setSubscriptionRetainedMinVersionId( + subscriptionRetentionBound.getRetainedMinVersionId()); + } } public void checkAndUpdateSearchIndex() { diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java index 0e4f41b9d8412..7cb5e7a63e038 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java @@ -29,6 +29,7 @@ import org.apache.iotdb.consensus.common.Peer; import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; import org.apache.iotdb.consensus.config.IoTConsensusConfig; +import org.apache.iotdb.consensus.exception.ConsensusGroupModifyPeerException; import org.apache.iotdb.consensus.i18n.IoTConsensusMessages; import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl; import org.apache.iotdb.consensus.iot.client.AsyncIoTConsensusServiceClient; @@ -170,13 +171,63 @@ public synchronized OptionalLong getMinFlushedSyncIndex() { return threads.stream().mapToLong(LogDispatcherThread::getLastFlushedSyncIndex).min(); } - public synchronized void checkAndFlushIndex() { - if (!threads.isEmpty()) { - threads.forEach( - thread -> { - IndexController controller = thread.getController(); - controller.update(controller.getCurrentIndex(), true); - }); + /** Wait for a fixed frontier, rather than chasing concurrent writes indefinitely. */ + public void waitUntilSyncLogCompleted(Peer peer, long targetIndex, long timeoutMs) + throws ConsensusGroupModifyPeerException { + final LogDispatcherThread thread; + synchronized (this) { + thread = + threads.stream() + .filter(t -> t.getPeer().equals(peer)) + .findFirst() + .orElseThrow( + () -> + new ConsensusGroupModifyPeerException( + String.format( + IoTConsensusMessages.EXCEPTION_NO_REPLICATION_CHANNEL_TO_ARG_F99CF90B, + peer))); + } + final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMs); + while (true) { + if (thread.replicationFailure != null || thread.stopped) { + throw new ConsensusGroupModifyPeerException( + String.format( + IoTConsensusMessages.EXCEPTION_REPLICATION_CHANNEL_TO_ARG_FAILED_402B2EC5, peer), + thread.replicationFailure); + } + if (thread.getCurrentSyncIndex() >= targetIndex) { + return; + } + if (System.nanoTime() >= deadline) { + throw new ConsensusGroupModifyPeerException( + String.format( + IoTConsensusMessages + .EXCEPTION_TIMED_OUT_WAITING_FOR_ARG_TO_APPLY_INDEX_ARG_ECAC0A5A, + peer, + targetIndex)); + } + try { + TimeUnit.MILLISECONDS.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new ConsensusGroupModifyPeerException( + String.format( + IoTConsensusMessages.EXCEPTION_INTERRUPTED_WAITING_FOR_REPLICATION_TO_ARG_64450D09, + peer), + e); + } + } + } + + public void checkAndFlushIndex() { + synchronized (this) { + if (!threads.isEmpty()) { + threads.forEach( + thread -> { + IndexController controller = thread.getController(); + controller.update(controller.getCurrentIndex(), true); + }); + } } // Single-replica regions do not have dispatcher threads, but still need periodic retention // recalculation for subscription-aware WAL cleanup. @@ -238,6 +289,7 @@ public class LogDispatcherThread implements Runnable { private static final long START_INDEX = 1; private final IoTConsensusConfig config; private final Peer peer; + private volatile Exception replicationFailure; private final IndexController controller; // A sliding window class that manages asynchronous pendingBatches private final SyncStatus syncStatus; @@ -412,6 +464,7 @@ public void run() { } catch (InterruptedException e) { Thread.currentThread().interrupt(); } catch (Exception e) { + replicationFailure = e; logger.error(IoTConsensusMessages.UNEXPECTED_ERROR_IN_LOG_DISPATCHER, peer, e); } runFinished.countDown(); @@ -483,9 +536,8 @@ public Batch getBatch() { // Prevents gap between logs. For example, some requests are not written into the queue when // the queue is full. In this case, requests need to be loaded from the WAL if (startIndex != prev.getSearchIndex()) { - boolean hasCorruptedData = - constructBatchFromWAL(startIndex, prev.getSearchIndex(), batches); - if (hasCorruptedData || !batches.canAccumulate()) { + constructBatchFromWAL(startIndex, prev.getSearchIndex(), batches); + if (!batches.canAccumulate()) { batches.buildIndex(); logger.debug( IoTConsensusMessages.ACCUMULATED_FROM_WAL, @@ -512,9 +564,8 @@ public Batch getBatch() { // Prevents gap between logs. For example, some logs are not written into the queue when // the queue is full. In this case, requests need to be loaded from the WAL if (current.getSearchIndex() != prev.getSearchIndex() + 1) { - boolean hasCorruptedData = - constructBatchFromWAL(prev.getSearchIndex() + 1, current.getSearchIndex(), batches); - if (hasCorruptedData || !batches.canAccumulate()) { + constructBatchFromWAL(prev.getSearchIndex() + 1, current.getSearchIndex(), batches); + if (!batches.canAccumulate()) { batches.buildIndex(); logger.debug( IoTConsensusMessages.ACCUMULATED_FROM_QUEUE_AND_WAL_GAP, @@ -608,14 +659,13 @@ public SyncStatus getSyncStatus() { return syncStatus; } - private boolean constructBatchFromWAL(long currentIndex, long maxIndex, Batch logBatches) { + private void constructBatchFromWAL(long currentIndex, long maxIndex, Batch logBatches) { logger.debug( IoTConsensusMessages.DATA_REGION_CONSTRUCT_FROM_WAL, peer.getGroupId().getId(), peer.getEndpoint().getIp(), currentIndex, maxIndex); - boolean hasCorruptedData = false; final boolean consensusGroupContainsUserData = impl.containsUserData(); // targetIndex is the index of request that we need to find long targetIndex = currentIndex; @@ -638,7 +688,13 @@ private boolean constructBatchFromWAL(long currentIndex, long maxIndex, Batch lo } else if (data.getSearchIndex() > targetIndex) { logger.warn( IoTConsensusMessages.SEARCH_ENTRY_FOUND_LARGER, targetIndex, data.getSearchIndex()); - hasCorruptedData = true; + throw new IllegalStateException( + String.format( + IoTConsensusMessages + .EXCEPTION_MISSING_WAL_ENTRIES_FOR_ARG_EXPECTED_ARG_FOUND_ARG_A9D194A5, + peer, + targetIndex, + data.getSearchIndex())); } targetIndex = data.getSearchIndex() + 1; // The WAL reader derives this bit directly from the entry type. Apply the group-level @@ -653,9 +709,6 @@ private boolean constructBatchFromWAL(long currentIndex, long maxIndex, Batch lo logEntry.setPhysicalTime(data.getPhysicalTime()); logBatches.addTLogEntry(logEntry, data.containsUserData()); } - // In the case of corrupt Data, we return true so that we can send a batch as soon as - // possible, avoiding potential duplication - return hasCorruptedData; } private void constructBatchIndexedFromConsensusRequest( diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java index 291a8d99d1b2e..41677b98700b6 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java @@ -26,12 +26,17 @@ import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType; import org.apache.iotdb.commons.request.IConsensusRequest; +import org.apache.iotdb.consensus.common.DataSet; import org.apache.iotdb.consensus.common.Peer; import org.apache.iotdb.consensus.common.request.DeserializedBatchIndexedConsensusRequest; import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; import org.apache.iotdb.consensus.config.IoTConsensusConfig; import org.apache.iotdb.consensus.config.UserDataTransferAuditClassifier; +import org.apache.iotdb.consensus.exception.ConsensusGroupModifyPeerException; import org.apache.iotdb.consensus.i18n.ConsensusMessages; +import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; +import org.apache.iotdb.consensus.iot.logdispatcher.LogDispatcher; +import org.apache.iotdb.consensus.iot.util.FakeConsensusReqReader; import org.apache.iotdb.consensus.iot.util.TestEntry; import org.apache.iotdb.consensus.iot.util.TestStateMachine; import org.apache.iotdb.rpc.TSStatusCode; @@ -53,11 +58,17 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; public class IoTConsensusServerImplTest { @@ -67,6 +78,90 @@ public class IoTConsensusServerImplTest { @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + @Test + public void testNewPeerImmediatelyProtectsWALAndCatchupTimesOut() throws Exception { + FakeConsensusReqReader reader = mock(FakeConsensusReqReader.class); + AtomicLong boundary = new AtomicLong(); + doAnswer( + invocation -> { + boundary.set(invocation.getArgument(0)); + return null; + }) + .when(reader) + .setSafelyDeletedSearchIndex(anyLong()); + ScheduledExecutorService background = Executors.newSingleThreadScheduledExecutor(); + Peer self = new Peer(new DataRegionId(1), 0, new TEndPoint("127.0.0.1", 6667)); + Peer peer = new Peer(new DataRegionId(1), 1, new TEndPoint("127.0.0.1", 6668)); + IoTConsensusServerImpl server = newRetentionTestServer(self, reader, background); + try { + server.getIndexObject().set(10); + server.checkAndUpdateSafeDeletedSearchIndex(); + assertEquals(Long.MAX_VALUE, boundary.get()); + server.buildSyncLogChannel(peer, false); + assertEquals(10, boundary.get()); + server.getLogDispatcher().waitUntilSyncLogCompleted(peer, 10, 100); + assertThrows( + ConsensusGroupModifyPeerException.class, + () -> server.getLogDispatcher().waitUntilSyncLogCompleted(peer, 11, 1)); + } finally { + server.stop(); + background.shutdownNow(); + } + } + + @Test + public void testMissingWALCannotProduceAnAcknowledgableBatch() throws Exception { + FakeConsensusReqReader reader = mock(FakeConsensusReqReader.class); + ConsensusReqReader.ReqIterator iterator = mock(ConsensusReqReader.ReqIterator.class); + when(reader.getReqIterator(anyLong())).thenReturn(iterator); + when(iterator.next()).thenReturn(new IndexedConsensusRequest(4, Collections.emptyList())); + ScheduledExecutorService background = Executors.newSingleThreadScheduledExecutor(); + Peer self = new Peer(new DataRegionId(1), 0, new TEndPoint("127.0.0.1", 6667)); + Peer peer = new Peer(new DataRegionId(1), 1, new TEndPoint("127.0.0.1", 6668)); + IoTConsensusServerImpl server = newRetentionTestServer(self, reader, background); + try { + server.getIndexObject().set(4); + server.buildSyncLogChannel(peer, 0, false); + Field threadsField = LogDispatcher.class.getDeclaredField("threads"); + threadsField.setAccessible(true); + @SuppressWarnings("unchecked") + List threads = + (List) threadsField.get(server.getLogDispatcher()); + LogDispatcher.LogDispatcherThread thread = threads.get(0); + assertThrows(IllegalStateException.class, thread::getBatch); + assertEquals(0, thread.getCurrentSyncIndex()); + // The real dispatcher loop must expose the failure to the add-peer waiter as well. + thread.run(); + assertThrows( + ConsensusGroupModifyPeerException.class, + () -> server.getLogDispatcher().waitUntilSyncLogCompleted(peer, 4, 100)); + } finally { + server.stop(); + background.shutdownNow(); + } + } + + private IoTConsensusServerImpl newRetentionTestServer( + Peer self, FakeConsensusReqReader reader, ScheduledExecutorService background) + throws Exception { + return new IoTConsensusServerImpl( + temporaryFolder.getRoot().getAbsolutePath(), + null, + DirectoryStrategyType.SEQUENCE_STRATEGY, + self, + Collections.singletonList(self), + new TestStateMachine() { + @Override + public synchronized DataSet read(IConsensusRequest request) { + return reader; + } + }, + background, + null, + null, + IoTConsensusConfig.newBuilder().build()); + } + @Test public void testSyncLogBecomesReadOnlyWhileWaiting() throws Exception { final ScheduledExecutorService backgroundTaskService = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java index b05beaf00c8a3..c28ed1dda2d2f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java @@ -120,6 +120,7 @@ public class WALNode implements IWALNode { // used to avoid write amplification caused by frequent snapshot private final Map memTableSnapshotCount = new ConcurrentHashMap<>(); // insert nodes whose search index are before this value can be deleted safely + private final Object walRetentionLock = new Object(); private volatile long safelyDeletedSearchIndex = DEFAULT_SAFELY_DELETED_SEARCH_INDEX; // notified whenever the value above advances, see setSafeDeletedSearchIndexListener private volatile LongConsumer safeDeletedSearchIndexListener; @@ -364,13 +365,14 @@ private class DeleteOutdatedFileTask implements Runnable { private File[] sortedWalFilesExcludingLast; private List activeOrPinnedMemTables; + private long cachedSafelyDeletedSearchIndex; + private int fileIndexAfterFilterSafelyDeleteIndex; private static final int MAX_RECURSION_TIME = 5; // the effective information ratio private double effectiveInfoRatio = 1.0d; - private int fileIndexAfterFilterSafelyDeleteIndex = Integer.MAX_VALUE; private List successfullyDeleted; private long deleteFileSize; @@ -395,7 +397,7 @@ private boolean initAndCheckIfNeedContinue() { this.sortedWalFilesExcludingLast = Arrays.copyOfRange(allWalFilesOfOneNode, 0, allWalFilesOfOneNode.length - 1); this.activeOrPinnedMemTables = checkpointManager.activeOrPinnedMemTables(); - this.fileIndexAfterFilterSafelyDeleteIndex = initFileIndexAfterFilterSafelyDeleteIndex(); + updateFileIndexAfterFilterSafelyDeleteIndex(); this.successfullyDeleted = new ArrayList<>(); this.deleteFileSize = 0; return true; @@ -486,18 +488,22 @@ private void deleteOutdatedFilesAndUpdateMetric() { File currentWal = sortedWalFilesExcludingLast[fileArrIdx]; WALFileStatus walFileStatus = WALFileUtils.parseStatusCode(currentWal.getName()); long versionId = WALFileUtils.parseVersionId(currentWal.getName()); - if (canDeleteFile(fileArrIdx, walFileStatus, versionId)) { - long fileSize = currentWal.length(); - if (org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(currentWal)) { - deleteFileSize += fileSize; - buffer.removeMemTableIdsOfWal(versionId); - successfullyDeleted.add(versionId); - } else { - logger.info( - StorageEngineMessages - .STORAGE_LOG_FAIL_TO_DELETE_OUTDATED_WAL_FILE_OF_WAL_NODE_1B1F2AF2, - currentWal, - identifier); + // A newly registered replica may lower the boundary while this cleanup is running. + // Check and delete under the same lock as boundary publication. + synchronized (walRetentionLock) { + if (canDeleteFile(fileArrIdx, walFileStatus, versionId)) { + long fileSize = currentWal.length(); + if (org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(currentWal)) { + deleteFileSize += fileSize; + buffer.removeMemTableIdsOfWal(versionId); + successfullyDeleted.add(versionId); + } else { + logger.info( + StorageEngineMessages + .STORAGE_LOG_FAIL_TO_DELETE_OUTDATED_WAL_FILE_OF_WAL_NODE_1B1F2AF2, + currentWal, + identifier); + } } } } @@ -506,11 +512,13 @@ private void deleteOutdatedFilesAndUpdateMetric() { removeDeletedWalFilesFromCache(successfullyDeleted); } - private int initFileIndexAfterFilterSafelyDeleteIndex() { - return safelyDeletedSearchIndex == DEFAULT_SAFELY_DELETED_SEARCH_INDEX - ? sortedWalFilesExcludingLast.length - : WALFileUtils.binarySearchFileBySearchIndex( - sortedWalFilesExcludingLast, safelyDeletedSearchIndex + 1); + private void updateFileIndexAfterFilterSafelyDeleteIndex() { + cachedSafelyDeletedSearchIndex = safelyDeletedSearchIndex; + fileIndexAfterFilterSafelyDeleteIndex = + cachedSafelyDeletedSearchIndex == DEFAULT_SAFELY_DELETED_SEARCH_INDEX + ? sortedWalFilesExcludingLast.length + : WALFileUtils.binarySearchFileBySearchIndex( + sortedWalFilesExcludingLast, cachedSafelyDeletedSearchIndex + 1); } /** Return true iff effective information ratio is too small or disk usage is too large. */ @@ -683,6 +691,10 @@ public boolean isContainsActiveOrPinnedMemTable(Long versionId) { } private boolean canDeleteFile(long fileArrIdx, WALFileStatus walFileStatus, long versionId) { + // Called under walRetentionLock: a cached result is reusable only for this boundary. + if (cachedSafelyDeletedSearchIndex != safelyDeletedSearchIndex) { + updateFileIndexAfterFilterSafelyDeleteIndex(); + } return (fileArrIdx < fileIndexAfterFilterSafelyDeleteIndex || walFileStatus == WALFileStatus.CONTAINS_NONE_SEARCH_INDEX) && versionId < subscriptionRetainedMinVersionId @@ -695,9 +707,15 @@ private boolean canDeleteFile(long fileArrIdx, WALFileStatus walFileStatus, long // region Search interfaces for consensus group @Override public void setSafelyDeletedSearchIndex(long safelyDeletedSearchIndex) { - this.safelyDeletedSearchIndex = safelyDeletedSearchIndex; + final boolean advanced; + synchronized (walRetentionLock) { + advanced = safelyDeletedSearchIndex > this.safelyDeletedSearchIndex; + this.safelyDeletedSearchIndex = safelyDeletedSearchIndex; + } + // Lowering retention for a new peer cannot make any additional LOAD directory reclaimable. + // In particular, do not run cleanup callbacks while its registration holds the write lock. final LongConsumer listener = this.safeDeletedSearchIndexListener; - if (listener != null) { + if (advanced && listener != null) { listener.accept(safelyDeletedSearchIndex); } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeTest.java index 1aedbbe742b7a..6d757a9b7f658 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeTest.java @@ -54,6 +54,8 @@ import java.io.File; import java.io.IOException; +import java.lang.reflect.Constructor; +import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; @@ -66,6 +68,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -331,6 +334,67 @@ public void testConcurrentCheckpoint() throws Exception { assertEquals(expectedMemTableId2Info, actualMemTableId2Info); } + @Test + public void testRetentionListenerOnlyRunsWhenBoundaryAdvances() { + AtomicInteger notifications = new AtomicInteger(); + walNode.setSafeDeletedSearchIndexListener(index -> notifications.incrementAndGet()); + walNode.setSafelyDeletedSearchIndex(0); + walNode.setSafelyDeletedSearchIndex(0); + assertEquals(0, notifications.get()); + walNode.setSafelyDeletedSearchIndex(1); + assertEquals(1, notifications.get()); + walNode.setSafelyDeletedSearchIndex(Long.MAX_VALUE); + assertEquals(2, notifications.get()); + walNode.setSafelyDeletedSearchIndex(0); + assertEquals(2, notifications.get()); + } + + @Test + public void testCleanupRechecksLoweredReplicationBoundary() throws Exception { + IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId); + walNode.onMemTableCreated(memTable, tsFilePathInLatestPartition(memTable)); + for (int index = 1; index <= 3; index++) { + InsertTabletNode node = getInsertTabletNode(devicePath, new long[] {index}); + node.setSearchIndex(index); + assertEquals( + WALFlushListener.Status.SUCCESS, + walNode + .log( + memTable.getMemTableId(), + node, + Collections.singletonList(new int[] {0, node.getRowCount()})) + .waitForResult()); + walNode.rollWALFile(); + } + walNode.onMemTableFlushed(memTable); + Awaitility.await().until(walNode::isAllWALEntriesConsumed); + walNode.setSafelyDeletedSearchIndex(Long.MAX_VALUE); + // Stop a real cleanup between its file-list snapshot and deletion, then register the lower + // replication boundary. No clock or scheduler timing is involved. + Class taskClass = Class.forName(WALNode.class.getName() + "$DeleteOutdatedFileTask"); + Constructor constructor = taskClass.getDeclaredConstructor(WALNode.class); + constructor.setAccessible(true); + Object task = constructor.newInstance(walNode); + Method initialize = taskClass.getDeclaredMethod("initAndCheckIfNeedContinue"); + initialize.setAccessible(true); + assertEquals(true, initialize.invoke(task)); + walNode.setSafelyDeletedSearchIndex(0); + Method delete = taskClass.getDeclaredMethod("deleteOutdatedFilesAndUpdateMetric"); + delete.setAccessible(true); + delete.invoke(task); + int retained = 0; + for (File file : WALFileUtils.listAllWALFiles(new File(logDirectory))) { + try (WALReader reader = new WALReader(file)) { + while (reader.hasNext()) { + if (reader.next().getValue() instanceof InsertTabletNode) { + retained++; + } + } + } + } + assertEquals(3, retained); + } + @Test public void testDeleteOutdatedFiles() throws Exception { List walFlushListeners = new ArrayList<>();