diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java index 80fd5302dabd4..8702749731080 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java @@ -47,6 +47,7 @@ import org.apache.iotdb.commons.conf.ConfigurationFileUtils; import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.conf.TrimProperties; +import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.enums.RepairDataPartitionTableProgressState; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; @@ -66,6 +67,7 @@ import org.apache.iotdb.commons.schema.tree.AlterTimeSeriesOperationType; import org.apache.iotdb.commons.schema.ttl.TTLCache; import org.apache.iotdb.commons.service.metric.MetricService; +import org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper; import org.apache.iotdb.commons.utils.AuthUtils; import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.commons.utils.StatusUtils; @@ -2679,19 +2681,60 @@ public TGetCommitProgressResp getCommitProgress(TGetCommitProgressReq req) { if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { return new TGetCommitProgressResp(status); } - final String keyPrefix = - req.getConsumerGroupId() + "##" + req.getTopicName() + "##" + req.getRegionId() + "##"; - final org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper keeper = + final CommitProgressKeeper keeper = subscriptionManager .getSubscriptionCoordinator() .getSubscriptionInfo() .getCommitProgressKeeper(); - final Map mergedWriterPositions = new LinkedHashMap<>(); + final RegionProgress mergedRegionProgress = + mergeCommitProgress( + keeper.getAllRegionProgress(), + req.getConsumerGroupId(), + req.getTopicName(), + req.getRegionId()); + final TGetCommitProgressResp resp = + new TGetCommitProgressResp(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + if (!mergedRegionProgress.getWriterPositions().isEmpty()) { + resp.setCommittedRegionProgress(serializeRegionProgress(mergedRegionProgress)); + } + return resp; + } - for (final Map.Entry entry : keeper.getAllRegionProgress().entrySet()) { - if (!entry.getKey().startsWith(keyPrefix)) { + static RegionProgress mergeCommitProgress( + final Map allRegionProgress, + final String consumerGroupId, + final String topicName, + final int regionId) { + final String regionIdString = new DataRegionId(regionId).toString(); + final Map mergedWriterPositions = new LinkedHashMap<>(); + final boolean hasVersionedProgress = + mergeCommitProgressForPrefix( + allRegionProgress, + CommitProgressKeeper.generateRegionKeyPrefix( + consumerGroupId, topicName, regionIdString), + mergedWriterPositions); + if (!hasVersionedProgress + || CommitProgressKeeper.isLegacyKeyUnambiguous( + consumerGroupId, topicName, regionIdString)) { + mergeCommitProgressForPrefix( + allRegionProgress, + CommitProgressKeeper.generateLegacyRegionKeyPrefix( + consumerGroupId, topicName, regionIdString), + mergedWriterPositions); + } + return new RegionProgress(mergedWriterPositions); + } + + private static boolean mergeCommitProgressForPrefix( + final Map allRegionProgress, + final String keyPrefix, + final Map mergedWriterPositions) { + boolean hasMatchingProgress = false; + for (final Map.Entry entry : allRegionProgress.entrySet()) { + if (!CommitProgressKeeper.isValidDataNodeProgressKey(entry.getKey(), keyPrefix)) { continue; } + hasMatchingProgress = true; final RegionProgress regionProgress = deserializeRegionProgress(entry.getValue()); if (Objects.isNull(regionProgress)) { continue; @@ -2705,13 +2748,7 @@ public TGetCommitProgressResp getCommitProgress(TGetCommitProgressReq req) { compareWriterProgress(newProgress, oldProgress) > 0 ? newProgress : oldProgress); } } - final TGetCommitProgressResp resp = - new TGetCommitProgressResp(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); - if (!mergedWriterPositions.isEmpty()) { - resp.setCommittedRegionProgress( - serializeRegionProgress(new RegionProgress(mergedWriterPositions))); - } - return resp; + return hasMatchingProgress; } private static RegionProgress deserializeRegionProgress(final ByteBuffer buffer) { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java index 823325514a92f..e503f59a21417 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java @@ -285,7 +285,7 @@ public void processLoadSnapshot(File snapshotDir) throws IOException { File snapshotFile = new File(snapshotDir, SNAPSHOT_FILENAME); if (!snapshotFile.exists()) { - // do nothing if the snapshot file is not existed + clear(); return; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java index fb0a89e71cabb..a6c710a0a1f74 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException; import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalJobExecutor; import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalPhantomReferenceCleaner; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; @@ -127,6 +128,11 @@ public void decreaseListenerReference(final PipeParameters parameters) regionListener.decreaseReference(parameters); } + public void reconcileListenerReferences(final Iterable pipeMetas) + throws IllegalPathException { + regionListener.reconcileReferences(pipeMetas); + } + /** Notify the region listener that the leader is ready to allow pipe operations. */ public void notifyLeaderReady() { regionListener.notifyLeaderReady(); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListener.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListener.java index 278e133494b50..6616e03939ac5 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListener.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListener.java @@ -20,6 +20,7 @@ package org.apache.iotdb.confignode.manager.pipe.agent.runtime; import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; import org.apache.iotdb.confignode.manager.pipe.source.ConfigRegionListeningFilter; import org.apache.iotdb.confignode.manager.pipe.source.ConfigRegionListeningQueue; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; @@ -57,6 +58,24 @@ public synchronized void decreaseReference(final PipeParameters parameters) } } + public synchronized void reconcileReferences(final Iterable pipeMetas) + throws IllegalPathException { + int referenceCount = 0; + for (final PipeMeta pipeMeta : pipeMetas) { + if (!ConfigRegionListeningFilter.parseListeningPlanTypeSet( + pipeMeta.getStaticMeta().getSourceParameters()) + .isEmpty()) { + referenceCount++; + } + } + listeningQueueReferenceCount = referenceCount; + if (referenceCount == 0) { + listeningQueue.close(); + } else { + listeningQueue.open(); + } + } + public synchronized boolean isLeaderReady() { return isLeaderReady.get(); } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java index 2dd49c1db36ed..772f46baa3162 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java @@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.common.rpc.thrift.TSchemaNode; import org.apache.iotdb.commons.auth.AuthException; +import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.schema.node.MNodeType; import org.apache.iotdb.commons.schema.ttl.TTLCache; @@ -804,10 +805,17 @@ public boolean loadSnapshot(final File latestSnapshotRootDir) { } }); if (result.get()) { - pipeInfo.getPipeTaskInfo().enrichPipeMetasWithRootUserForCompatibility(); - LOGGER.info( - ConfigNodeMessages.CONFIGNODESNAPSHOT_LOAD_SNAPSHOT_SUCCESS_LATESTSNAPSHOTROOTDIR, - latestSnapshotRootDir); + try { + PipeConfigNodeAgent.runtime() + .reconcileListenerReferences(pipeInfo.getPipeTaskInfo().getPipeMetaList()); + pipeInfo.getPipeTaskInfo().enrichPipeMetasWithRootUserForCompatibility(); + LOGGER.info( + ConfigNodeMessages.CONFIGNODESNAPSHOT_LOAD_SNAPSHOT_SUCCESS_LATESTSNAPSHOTROOTDIR, + latestSnapshotRootDir); + } catch (final IllegalPathException e) { + result.set(false); + LOGGER.error(ConfigNodeMessages.LOAD_SNAPSHOT_ERROR, e); + } } // Propagate any snapshot-load failure so callers (e.g. the AddPeer flow) do not treat a // partially or wholly failed load as success. diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeInfo.java index aec8c70777031..6e6d5fcbcb0c3 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeInfo.java @@ -311,11 +311,6 @@ public void processLoadSnapshot(final File snapshotDir) throws IOException { try { pipeTaskInfo.processLoadSnapshot(snapshotDir); - - for (final PipeMeta pipeMeta : pipeTaskInfo.getPipeMetaList()) { - PipeConfigNodeAgent.runtime() - .increaseListenerReference(pipeMeta.getStaticMeta().getSourceParameters()); - } } catch (final Exception ex) { LOGGER.error(ConfigNodeMessages.FAILED_TO_LOAD_PIPE_TASK_INFO_FROM_SNAPSHOT, ex); loadPipeTaskInfoException = ex; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/quota/QuotaInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/quota/QuotaInfo.java index 783ad6c336891..1dbb006896cab 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/quota/QuotaInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/quota/QuotaInfo.java @@ -259,6 +259,7 @@ public Map getThrottleQuotaLimit() { public void clear() { spaceQuotaLimit.clear(); + spaceQuotaUsage.clear(); throttleQuotaLimit.clear(); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/TemplatePreSetTable.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/TemplatePreSetTable.java index 9105e5a7d01da..985e005e967ab 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/TemplatePreSetTable.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/TemplatePreSetTable.java @@ -149,6 +149,8 @@ public void processLoadSnapshot(File snapshotDir) throws IOException { try { File snapshotFile = new File(snapshotDir, SNAPSHOT_FILENAME); if (!snapshotFile.exists()) { + // Empty preset tables are represented by the absence of a snapshot file. + templatePreSetMap.clear(); return; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/TemplateTable.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/TemplateTable.java index 10fa66108bb79..29878b47dfeb6 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/TemplateTable.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/TemplateTable.java @@ -262,9 +262,8 @@ public void processLoadSnapshot(File snapshotDir) throws IOException { BufferedInputStream bufferedInputStream = new BufferedInputStream(fileInputStream)) { // Load snapshot of template this.templateMap.clear(); + this.templateIdMap.clear(); deserialize(bufferedInputStream); - bufferedInputStream.close(); - fileInputStream.close(); } finally { templateReadWriteLock.writeLock().unlock(); } @@ -272,6 +271,13 @@ public void processLoadSnapshot(File snapshotDir) throws IOException { @TestOnly public void clear() { - this.templateMap.clear(); + templateReadWriteLock.writeLock().lock(); + try { + this.templateMap.clear(); + this.templateIdMap.clear(); + this.templateIdGenerator.set(0); + } finally { + templateReadWriteLock.writeLock().unlock(); + } } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java index fad083b69edd1..df6a0d5123120 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java @@ -864,9 +864,19 @@ public boolean isTopicSubscribedByConsumerGroup( public TSStatus alterConsumerGroup(AlterConsumerGroupPlan plan) { acquireWriteLock(); try { - ConsumerGroupMeta consumerGroupMeta = plan.getConsumerGroupMeta(); + final ConsumerGroupMeta consumerGroupMeta = plan.getConsumerGroupMeta(); if (Objects.nonNull(consumerGroupMeta)) { - String consumerGroupId = consumerGroupMeta.getConsumerGroupId(); + final String consumerGroupId = consumerGroupMeta.getConsumerGroupId(); + final ConsumerGroupMeta currentConsumerGroupMeta = + consumerGroupMetaKeeper.containsConsumerGroupMeta(consumerGroupId) + ? consumerGroupMetaKeeper.getConsumerGroupMeta(consumerGroupId) + : null; + if (Objects.nonNull(currentConsumerGroupMeta)) { + ConsumerGroupMeta.getTopicsUnsubByGroup(currentConsumerGroupMeta, consumerGroupMeta) + .forEach( + topicName -> + commitProgressKeeper.removeTopicProgress(consumerGroupId, topicName)); + } consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupId); if (!consumerGroupMeta.isEmpty()) { consumerGroupMetaKeeper.addConsumerGroupMeta(consumerGroupId, consumerGroupMeta); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ConfigManagerCommitProgressTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ConfigManagerCommitProgressTest.java new file mode 100644 index 0000000000000..39707921a7ca0 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ConfigManagerCommitProgressTest.java @@ -0,0 +1,152 @@ +/* + * 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.manager; + +import org.apache.iotdb.commons.consensus.DataRegionId; +import org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper; +import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress; +import org.apache.iotdb.rpc.subscription.payload.poll.WriterId; +import org.apache.iotdb.rpc.subscription.payload.poll.WriterProgress; + +import org.junit.Test; + +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.nio.ByteBuffer; +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.junit.Assert.assertEquals; + +public class ConfigManagerCommitProgressTest { + + @Test + public void testVersionedProgressMergesLegacyProgressDuringRollingUpgrade() throws Exception { + final int regionId = 1; + final String regionIdString = new DataRegionId(regionId).toString(); + final WriterId versionedWriter = new WriterId(regionIdString, 1); + final WriterId legacyWriter = new WriterId(regionIdString, 2); + final RegionProgress versionedProgress = + createRegionProgress(versionedWriter, new WriterProgress(100L, 1L)); + final RegionProgress legacyProgress = + createRegionProgress(legacyWriter, new WriterProgress(200L, 2L)); + final Map progressMap = new LinkedHashMap<>(); + progressMap.put( + CommitProgressKeeper.generateLegacyKey("cg", "topic", regionIdString, 1), + serialize(legacyProgress)); + progressMap.put( + CommitProgressKeeper.generateKey("cg", "topic", regionIdString, 2), + serialize(versionedProgress)); + final Map expectedWriterPositions = new LinkedHashMap<>(); + expectedWriterPositions.put(versionedWriter, new WriterProgress(100L, 1L)); + expectedWriterPositions.put(legacyWriter, new WriterProgress(200L, 2L)); + + assertEquals( + new RegionProgress(expectedWriterPositions), + ConfigManager.mergeCommitProgress(progressMap, "cg", "topic", regionId)); + } + + @Test + public void testLegacyProgressIsUsedWhenVersionedProgressIsAbsent() throws Exception { + final int regionId = 2; + final String regionIdString = new DataRegionId(regionId).toString(); + final WriterId writerId = new WriterId(regionIdString, 1); + final RegionProgress legacyProgress = + createRegionProgress(writerId, new WriterProgress(100L, 1L)); + final Map progressMap = new LinkedHashMap<>(); + progressMap.put( + CommitProgressKeeper.generateLegacyKey("cg", "topic", regionIdString, 1), + serialize(legacyProgress)); + + assertEquals( + legacyProgress, ConfigManager.mergeCommitProgress(progressMap, "cg", "topic", regionId)); + } + + @Test + public void testLegacyKeyCannotMasqueradeAsVersionedProgress() throws Exception { + final int regionId = 3; + final String regionIdString = new DataRegionId(regionId).toString(); + final WriterId legacyWriter = new WriterId(regionIdString, 1); + final WriterId masqueradingWriter = new WriterId(regionIdString, 2); + final RegionProgress legacyProgress = + createRegionProgress(legacyWriter, new WriterProgress(100L, 1L)); + final RegionProgress masqueradingProgress = + createRegionProgress(masqueradingWriter, new WriterProgress(200L, 2L)); + final Map progressMap = new LinkedHashMap<>(); + progressMap.put( + CommitProgressKeeper.generateLegacyKey("cg", "topic", regionIdString, 1), + serialize(legacyProgress)); + progressMap.put( + CommitProgressKeeper.generateLegacyKey( + CommitProgressKeeper.generateRegionKey("cg", "topic", regionIdString), + "legacyTopic", + "legacyRegion", + 2), + serialize(masqueradingProgress)); + + assertEquals( + legacyProgress, ConfigManager.mergeCommitProgress(progressMap, "cg", "topic", regionId)); + } + + @Test + public void testVersionedProgressSeparatesLegacyKeyCollisions() throws Exception { + final int regionId = 4; + final String regionIdString = new DataRegionId(regionId).toString(); + final WriterId firstWriter = new WriterId(regionIdString, 1); + final WriterId secondWriter = new WriterId(regionIdString, 2); + final WriterId legacyWriter = new WriterId(regionIdString, 3); + final RegionProgress firstProgress = + createRegionProgress(firstWriter, new WriterProgress(100L, 1L)); + final RegionProgress secondProgress = + createRegionProgress(secondWriter, new WriterProgress(200L, 2L)); + final RegionProgress legacyProgress = + createRegionProgress(legacyWriter, new WriterProgress(300L, 3L)); + final Map progressMap = new LinkedHashMap<>(); + progressMap.put( + CommitProgressKeeper.generateKey("a##b", "c", regionIdString, 1), serialize(firstProgress)); + progressMap.put( + CommitProgressKeeper.generateKey("a", "b##c", regionIdString, 2), + serialize(secondProgress)); + progressMap.put( + CommitProgressKeeper.generateLegacyKey("a##b", "c", regionIdString, 3), + serialize(legacyProgress)); + + assertEquals( + firstProgress, ConfigManager.mergeCommitProgress(progressMap, "a##b", "c", regionId)); + assertEquals( + secondProgress, ConfigManager.mergeCommitProgress(progressMap, "a", "b##c", regionId)); + } + + private static RegionProgress createRegionProgress( + final WriterId writerId, final WriterProgress writerProgress) { + final Map writerPositions = new LinkedHashMap<>(); + writerPositions.put(writerId, writerProgress); + return new RegionProgress(writerPositions); + } + + private static ByteBuffer serialize(final RegionProgress regionProgress) throws Exception { + try (final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + final DataOutputStream dos = new DataOutputStream(baos)) { + regionProgress.serialize(dos); + dos.flush(); + return ByteBuffer.wrap(baos.toByteArray()).asReadOnlyBuffer(); + } + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListenerSnapshotTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListenerSnapshotTest.java new file mode 100644 index 0000000000000..1ac8cf4fa18ea --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListenerSnapshotTest.java @@ -0,0 +1,59 @@ +/* + * 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.manager.pipe.agent.runtime; + +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; +import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant; + +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +public class PipeConfigRegionListenerSnapshotTest { + + @Test + public void testReconcileReferencesReplacesSnapshotState() throws Exception { + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put(PipeSourceConstant.EXTRACTOR_INCLUSION_KEY, "schema.database.create"); + final PipeStaticMeta staticMeta = + new PipeStaticMeta( + "pipe", 1, sourceAttributes, Collections.emptyMap(), Collections.emptyMap()); + final PipeMeta pipeMeta = Mockito.mock(PipeMeta.class); + Mockito.when(pipeMeta.getStaticMeta()).thenReturn(staticMeta); + + final PipeConfigRegionListener listener = new PipeConfigRegionListener(); + listener.increaseReference(staticMeta.getSourceParameters()); + listener.listener().close(); + + listener.reconcileReferences(Collections.emptyList()); + listener.increaseReference(staticMeta.getSourceParameters()); + Assert.assertTrue(listener.listener().isOpened()); + + listener.reconcileReferences(Collections.singletonList(pipeMeta)); + listener.reconcileReferences(Collections.singletonList(pipeMeta)); + listener.decreaseReference(staticMeta.getSourceParameters()); + Assert.assertFalse(listener.listener().isOpened()); + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/ExternalServiceInfoTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/ExternalServiceInfoTest.java index 205c957192455..964413e4d62ac 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/ExternalServiceInfoTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/ExternalServiceInfoTest.java @@ -90,4 +90,25 @@ public void testSnapshot() throws TException, IOException, IllegalPathException serviceInfo.getRawDatanodeToServiceInfos(), serviceInfoBefore.getRawDatanodeToServiceInfos()); } + + @Test + public void testLoadEmptySnapshotClearsExistingState() throws Exception { + final File emptySnapshotDir = new File(BASE_OUTPUT_PATH, "empty-external-service-snapshot"); + final ExternalServiceInfo emptySnapshotSource = new ExternalServiceInfo(); + final ExternalServiceInfo target = new ExternalServiceInfo(); + try { + Assert.assertTrue(emptySnapshotDir.mkdirs() || emptySnapshotDir.isDirectory()); + target.addService( + new CreateExternalServicePlan( + 1, new ServiceInfo("TEST", "testClassName", ServiceInfo.ServiceType.USER_DEFINED))); + Assert.assertFalse(target.getRawDatanodeToServiceInfos().isEmpty()); + + Assert.assertTrue(emptySnapshotSource.processTakeSnapshot(emptySnapshotDir)); + target.processLoadSnapshot(emptySnapshotDir); + + Assert.assertTrue(target.getRawDatanodeToServiceInfos().isEmpty()); + } finally { + FileUtils.deleteDirectory(emptySnapshotDir); + } + } } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/QuotaInfoTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/QuotaInfoTest.java index 88f06f602e25f..a568e842752bb 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/QuotaInfoTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/QuotaInfoTest.java @@ -37,6 +37,7 @@ import java.io.File; import java.io.IOException; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -97,6 +98,19 @@ public void testSnapshot() throws TException, IOException { QuotaInfo quotaInfo2 = new QuotaInfo(); quotaInfo2.processLoadSnapshot(snapshotDir); + final TSpaceQuota staleQuota = new TSpaceQuota(); + staleQuota.setDeviceNum(1); + staleQuota.setTimeserieNum(1); + staleQuota.setDiskSize(1); + quotaInfo2.setSpaceQuota( + new SetSpaceQuotaPlan(Collections.singletonList("root.stale"), staleQuota)); + Assert.assertTrue(quotaInfo2.getSpaceQuotaUsage().containsKey("root.stale")); + + quotaInfo2.processLoadSnapshot(snapshotDir); + + Assert.assertFalse(quotaInfo2.getSpaceQuotaLimit().containsKey("root.stale")); + Assert.assertFalse(quotaInfo2.getSpaceQuotaUsage().containsKey("root.stale")); + Assert.assertEquals(quotaInfo.getSpaceQuotaLimit(), quotaInfo2.getSpaceQuotaLimit()); Assert.assertEquals(quotaInfo.getThrottleQuotaLimit(), quotaInfo2.getThrottleQuotaLimit()); } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TemplatePreSetTableTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TemplatePreSetTableTest.java index fd142122c2b55..60ae269b096ff 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TemplatePreSetTableTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TemplatePreSetTableTest.java @@ -104,11 +104,24 @@ public void testSnapshot() throws IllegalPathException { newTemplatePreSetTable = new TemplatePreSetTable(); newTemplatePreSetTable.processLoadSnapshot(snapshotDir); - Assert.assertTrue(templatePreSetTable.isPreSet(templateId1, templateSetPath1)); - Assert.assertTrue(templatePreSetTable.isPreSet(templateId2, templateSetPath1)); - Assert.assertTrue(templatePreSetTable.isPreSet(templateId2, templateSetPath2)); + Assert.assertTrue(newTemplatePreSetTable.isPreSet(templateId1, templateSetPath1)); + Assert.assertTrue(newTemplatePreSetTable.isPreSet(templateId2, templateSetPath1)); + Assert.assertTrue(newTemplatePreSetTable.isPreSet(templateId2, templateSetPath2)); } catch (IOException e) { Assert.fail(); } } + + @Test + public void testEmptySnapshotReplacesExistingState() throws IOException, IllegalPathException { + final int templateId = 5; + final PartialPath templateSetPath = new PartialPath("root.db.t1"); + + Assert.assertTrue(templatePreSetTable.processTakeSnapshot(snapshotDir)); + templatePreSetTable.preSetTemplate(templateId, templateSetPath); + + templatePreSetTable.processLoadSnapshot(snapshotDir); + + Assert.assertFalse(templatePreSetTable.isPreSet(templateId, templateSetPath)); + } } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TemplateTableTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TemplateTableTest.java index dd5bb5d8bb8b3..d67edf5ace602 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TemplateTableTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TemplateTableTest.java @@ -77,7 +77,8 @@ public void testSnapshot() throws IOException, MetadataException { } templateTable.processTakeSnapshot(snapshotDir); - templateTable.clear(); + Template staleTemplate = newSchemaTemplate("stale_template"); + templateTable.createTemplate(staleTemplate); templateTable.processLoadSnapshot(snapshotDir); // show nodes in schemaengine template @@ -86,6 +87,13 @@ public void testSnapshot() throws IOException, MetadataException { Template template = templates.get(i); Assert.assertEquals(template, templateTable.getTemplate(templateNameTmp)); } + + try { + templateTable.getTemplate(staleTemplate.getId()); + Assert.fail("Template created after the snapshot should be removed when loading it"); + } catch (MetadataException expected) { + // expected + } } @Test diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java index 9b86f1d402af0..f87afe129e5a8 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java @@ -20,7 +20,11 @@ package org.apache.iotdb.confignode.persistence.subscription; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper; +import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta; +import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta; import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; +import org.apache.iotdb.confignode.consensus.request.write.subscription.consumer.AlterConsumerGroupPlan; import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.AlterTopicPlan; import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.CreateTopicPlan; import org.apache.iotdb.confignode.consensus.response.subscription.TopicTableResp; @@ -33,10 +37,12 @@ import org.junit.Assert; import org.junit.Test; +import java.nio.ByteBuffer; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.UUID; public class SubscriptionInfoTest { @@ -171,6 +177,40 @@ public void testAlterTopicOwnerAndShowTopicOwner() { Assert.assertFalse(showTopicInfo.getTopicAttributes().contains("owner-lease-expire-time-ms")); } + @Test + public void testAlterConsumerGroupRemovesUnsubscribedTopicProgress() { + final String consumerGroupId = "consumer-group"; + final String consumerId = "consumer"; + final String removedTopic = "removed-topic"; + final String retainedTopic = "retained-topic"; + final String regionId = "DataRegion[1]"; + final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); + final ConsumerGroupMeta currentMeta = + new ConsumerGroupMeta( + consumerGroupId, 1, new ConsumerMeta(consumerId, 1, Collections.emptyMap())); + currentMeta.addSubscription(consumerId, Set.of(removedTopic, retainedTopic)); + subscriptionInfo.alterConsumerGroup(new AlterConsumerGroupPlan(currentMeta)); + + final CommitProgressKeeper keeper = subscriptionInfo.getCommitProgressKeeper(); + final String removedVersionedKey = + CommitProgressKeeper.generateKey(consumerGroupId, removedTopic, regionId, 1); + final String removedLegacyKey = + CommitProgressKeeper.generateLegacyKey(consumerGroupId, removedTopic, regionId, 2); + final String retainedKey = + CommitProgressKeeper.generateKey(consumerGroupId, retainedTopic, regionId, 1); + keeper.updateRegionProgress(removedVersionedKey, ByteBuffer.wrap(new byte[] {1})); + keeper.updateRegionProgress(removedLegacyKey, ByteBuffer.wrap(new byte[] {2})); + keeper.updateRegionProgress(retainedKey, ByteBuffer.wrap(new byte[] {3})); + + final ConsumerGroupMeta updatedMeta = currentMeta.deepCopy(); + updatedMeta.removeSubscription(consumerId, Collections.singleton(removedTopic)); + subscriptionInfo.alterConsumerGroup(new AlterConsumerGroupPlan(updatedMeta)); + + Assert.assertNull(keeper.getRegionProgress(removedVersionedKey)); + Assert.assertNull(keeper.getRegionProgress(removedLegacyKey)); + Assert.assertNotNull(keeper.getRegionProgress(retainedKey)); + } + private TopicMeta createTopicMeta( final String topicName, final String ownerId, final long ownerEpoch) { final Map topicAttributes = new HashMap<>(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java index 0c2b0fc53510f..59493ef33271b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java @@ -31,6 +31,7 @@ import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; import org.apache.tsfile.exception.write.WriteProcessException; +import org.apache.tsfile.external.commons.io.FileUtils; import org.apache.tsfile.utils.Pair; import org.apache.tsfile.write.record.Tablet; import org.slf4j.Logger; @@ -237,13 +238,25 @@ public synchronized List> sealTsFiles() } final List> list = new ArrayList<>(); - if (!treeModeTsFileBuilder.isEmpty()) { - list.addAll(treeModeTsFileBuilder.convertTabletToTsFileWithDBInfo()); - } - if (!tableModeTsFileBuilder.isEmpty()) { - list.addAll(tableModeTsFileBuilder.convertTabletToTsFileWithDBInfo()); + boolean sealedSuccessfully = false; + try { + if (!treeModeTsFileBuilder.isEmpty()) { + list.addAll(treeModeTsFileBuilder.convertTabletToTsFileWithDBInfo()); + } + if (!tableModeTsFileBuilder.isEmpty()) { + list.addAll(tableModeTsFileBuilder.convertTabletToTsFileWithDBInfo()); + } + sealedSuccessfully = true; + return list; + } finally { + if (!sealedSuccessfully) { + for (final Pair sealedFile : list) { + if (sealedFile.right.exists() && !FileUtils.deleteQuietly(sealedFile.right)) { + LOGGER.warn(DataNodePipeMessages.FAILED_TO_DELETE_BATCH_FILE_THIS_FILE, sealedFile); + } + } + } } - return list; } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java index 8689069334ea9..a535f5bb33a30 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java @@ -279,19 +279,24 @@ private void doTransfer( final List> dbTsFilePairs = batchToTransfer.sealTsFiles(); final Map, Double> pipe2WeightMap = batchToTransfer.deepCopyPipe2WeightMap(); - for (final Pair dbTsFile : dbTsFilePairs) { - doTransfer( - pipe2WeightMap, socket, dbTsFile.right, null, dbTsFile.left, dbTsFile.right.getName()); - try { - RetryUtils.retryOnException( - () -> { - FileUtils.delete(dbTsFile.right); - return null; - }); - } catch (final NoSuchFileException e) { - LOGGER.info(DataNodePipeMessages.THE_FILE_IS_NOT_FOUND_MAY_ALREADY, dbTsFile); - } catch (final Exception e) { - LOGGER.warn(DataNodePipeMessages.FAILED_TO_DELETE_BATCH_FILE_THIS_FILE, dbTsFile); + try { + for (final Pair dbTsFile : dbTsFilePairs) { + doTransfer( + pipe2WeightMap, socket, dbTsFile.right, null, dbTsFile.left, dbTsFile.right.getName()); + } + } finally { + for (final Pair dbTsFile : dbTsFilePairs) { + try { + RetryUtils.retryOnException( + () -> { + FileUtils.delete(dbTsFile.right); + return null; + }); + } catch (final NoSuchFileException e) { + LOGGER.info(DataNodePipeMessages.THE_FILE_IS_NOT_FOUND_MAY_ALREADY, dbTsFile); + } catch (final Exception e) { + LOGGER.warn(DataNodePipeMessages.FAILED_TO_DELETE_BATCH_FILE_THIS_FILE, dbTsFile); + } } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java index 213e43ab0b844..d3270c3a4acb4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java @@ -70,6 +70,7 @@ import com.google.common.collect.ImmutableSet; import org.apache.tsfile.exception.write.WriteProcessException; +import org.apache.tsfile.external.commons.io.FileUtils; import org.apache.tsfile.utils.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -262,6 +263,7 @@ private void transferInBatchWithoutCheck( final AtomicInteger eventsReferenceCount = new AtomicInteger(dbTsFilePairs.size()); final AtomicBoolean eventsHadBeenAddedToRetryQueue = new AtomicBoolean(false); + int transferredFileCount = 0; try { for (final Pair sealedFile : dbTsFilePairs) { transfer( @@ -275,8 +277,17 @@ private void transferInBatchWithoutCheck( null, false, sealedFile.left)); + transferredFileCount++; } } catch (final Exception e) { + for (int i = transferredFileCount; i < dbTsFilePairs.size(); i++) { + final Pair untransferredFile = dbTsFilePairs.get(i); + if (untransferredFile.right.exists() + && !FileUtils.deleteQuietly(untransferredFile.right)) { + LOGGER.warn( + DataNodePipeMessages.FAILED_TO_DELETE_BATCH_FILE_THIS_FILE, untransferredFile); + } + } PipeLogger.log( ignored -> LOGGER.warn(DataNodePipeMessages.FAILED_TO_TRANSFER_TSFILE_BATCH, dbTsFilePairs, e), @@ -457,24 +468,30 @@ private boolean transferWithoutCheck(final TsFileInsertionEvent tsFileInsertionE private void transfer(final PipeTransferTsFileHandler pipeTransferTsFileHandler) { transferTsFileCounter.incrementAndGet(); - CompletableFuture completableFuture = - CompletableFuture.supplyAsync( - () -> { - AsyncPipeDataTransferServiceClient client = null; - try { - client = transferTsFileClientManager.borrowClient(); - markHandshakeSucceeded(); - pipeTransferTsFileHandler.transfer(transferTsFileClientManager, client); - } catch (final Exception ex) { - markSchedulingDelayIfHandshakeFailed(client); - logOnClientException(client, ex); - pipeTransferTsFileHandler.onError(ex); - } finally { - transferTsFileCounter.decrementAndGet(); - } - return null; - }, - transferTsFileClientManager.getExecutor()); + final CompletableFuture completableFuture; + try { + completableFuture = + CompletableFuture.supplyAsync( + () -> { + AsyncPipeDataTransferServiceClient client = null; + try { + client = transferTsFileClientManager.borrowClient(); + markHandshakeSucceeded(); + pipeTransferTsFileHandler.transfer(transferTsFileClientManager, client); + } catch (final Exception ex) { + markSchedulingDelayIfHandshakeFailed(client); + logOnClientException(client, ex); + pipeTransferTsFileHandler.onError(ex); + } finally { + transferTsFileCounter.decrementAndGet(); + } + return null; + }, + transferTsFileClientManager.getExecutor()); + } catch (final RuntimeException e) { + transferTsFileCounter.decrementAndGet(); + throw e; + } if (PipeConfig.getInstance().isTransferTsFileSync()) { try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java index 4c25d03de5308..7f06b771a0cb4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java @@ -38,6 +38,7 @@ import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceWithModReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFileSealWithModReq; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; +import org.apache.iotdb.pipe.api.exception.PipeConnectionException; import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq; @@ -159,6 +160,8 @@ public void transfer( DataNodePipeMessages.CLIENT_HAS_BEEN_RETURNED_TO_THE_POOL, sink.isClosed() ? "CLOSED" : "NOT CLOSED", tsFile); + onError( + new PipeConnectionException(DataNodePipeMessages.CLIENT_HAS_BEEN_RETURNED_TO_THE_POOL)); return; } @@ -476,8 +479,26 @@ public void clearEventsReferenceCount() { @Override public void close() { - super.close(); - releaseReadBufferMemoryBlock(); + try { + if (reader != null) { + reader.close(); + reader = null; + } + + if (currentFile.exists() + && events.stream().anyMatch(event -> !(event instanceof PipeTsFileInsertionEvent))) { + RetryUtils.retryOnException( + () -> { + FileUtils.delete(currentFile); + return null; + }); + } + } catch (final IOException e) { + LOGGER.warn(DataNodePipeMessages.FAILED_TO_CLOSE_FILE_READER_OR_DELETE, e); + } finally { + super.close(); + releaseReadBufferMemoryBlock(); + } } private void releaseReadBufferMemoryBlock() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java index d13b9664f8ab2..46d4034cf35f6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java @@ -351,18 +351,23 @@ private void doTransfer(final PipeTabletEventTsFileBatch batchToTransfer) final List> dbTsFilePairs = batchToTransfer.sealTsFiles(); final Map, Double> pipe2WeightMap = batchToTransfer.deepCopyPipe2WeightMap(); - for (final Pair dbTsFile : dbTsFilePairs) { - doTransfer(pipe2WeightMap, dbTsFile.right, null, dbTsFile.left); - try { - RetryUtils.retryOnException( - () -> { - FileUtils.delete(dbTsFile.right); - return null; - }); - } catch (final NoSuchFileException e) { - LOGGER.info(DataNodePipeMessages.THE_FILE_IS_NOT_FOUND_MAY_ALREADY, dbTsFile); - } catch (final Exception e) { - LOGGER.warn(DataNodePipeMessages.FAILED_TO_DELETE_BATCH_FILE_THIS_FILE, dbTsFile); + try { + for (final Pair dbTsFile : dbTsFilePairs) { + doTransfer(pipe2WeightMap, dbTsFile.right, null, dbTsFile.left); + } + } finally { + for (final Pair dbTsFile : dbTsFilePairs) { + try { + RetryUtils.retryOnException( + () -> { + FileUtils.delete(dbTsFile.right); + return null; + }); + } catch (final NoSuchFileException e) { + LOGGER.info(DataNodePipeMessages.THE_FILE_IS_NOT_FOUND_MAY_ALREADY, dbTsFile); + } catch (final Exception e) { + LOGGER.warn(DataNodePipeMessages.FAILED_TO_DELETE_BATCH_FILE_THIS_FILE, dbTsFile); + } } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilder.java index bcb7940888c85..ce7e81e348125 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilder.java @@ -83,10 +83,15 @@ public List> convertTabletToTsFileWithDBInfo() throws IOExcep return new ArrayList<>(0); } final List> pairList = new ArrayList<>(); - for (Map.Entry> entry : dataBase2TabletList.entrySet()) { - pairList.addAll(writeTableModelTabletsToTsFiles(entry.getValue(), entry.getKey())); + try { + for (Map.Entry> entry : dataBase2TabletList.entrySet()) { + pairList.addAll(writeTableModelTabletsToTsFiles(entry.getValue(), entry.getKey())); + } + return pairList; + } catch (final IOException | RuntimeException e) { + pairList.forEach(pair -> FileUtils.deleteQuietly(pair.right)); + throw e; } - return pairList; } @Override @@ -150,7 +155,13 @@ List> writeTableModelTabletsToTsFiles( // Try making the tsfile size as large as possible while (!device2TabletsLinkedList.isEmpty()) { if (Objects.isNull(fileWriter)) { - fileWriter = new TsFileWriter(createFile()); + final File file = createFile(); + try { + fileWriter = new TsFileWriter(file); + } catch (final IOException | RuntimeException e) { + FileUtils.deleteQuietly(file); + throw e; + } } try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java index fb275a1893fed..eb20a33e14785 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java @@ -32,6 +32,7 @@ import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.exception.write.WriteProcessException; +import org.apache.tsfile.external.commons.io.FileUtils; import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.utils.DateUtils; @@ -79,6 +80,7 @@ public PipeTableModelTsFileBuilderV2( @Override public void bufferTableModelTablet(String dataBase, Tablet tablet) { dataBase2TabletList.computeIfAbsent(dataBase, db -> new ArrayList<>()).add(tablet); + fallbackBuilder.bufferTableModelTablet(dataBase, tablet); } @Override @@ -92,13 +94,14 @@ public List> convertTabletToTsFileWithDBInfo() throws IOExcep if (dataBase2TabletList.isEmpty()) { return new ArrayList<>(0); } + final List> pairList = new ArrayList<>(); try { - final List> pairList = new ArrayList<>(); for (final String dataBase : dataBase2TabletList.keySet()) { pairList.addAll(writeTabletsToTsFiles(dataBase)); } return pairList; } catch (final Exception e) { + pairList.forEach(pair -> FileUtils.deleteQuietly(pair.right)); LOGGER.warn( DataNodePipeMessages .EXCEPTION_OCCURRED_WHEN_PIPETABLEMODELTSFILEBUILDERV2_WRITING_TABLETS_TO, @@ -131,10 +134,15 @@ private List> writeTabletsToTsFiles(final String dataBase) throws WriteProcessException { final IMemTable memTable = new PrimitiveMemTable(null, null); final List> sealedFiles = new ArrayList<>(); - try (final RestorableTsFileIOWriter writer = new RestorableTsFileIOWriter(createFile())) { - writeTabletsIntoOneFile(dataBase, memTable, writer); - sealedFiles.add(new Pair<>(dataBase, writer.getFile())); + File file = null; + try { + file = createFile(); + try (final RestorableTsFileIOWriter writer = new RestorableTsFileIOWriter(file)) { + writeTabletsIntoOneFile(dataBase, memTable, writer); + sealedFiles.add(new Pair<>(dataBase, writer.getFile())); + } } catch (final Exception e) { + FileUtils.deleteQuietly(file); LOGGER.warn( DataNodePipeMessages.BATCH_ID_FAILED_TO_WRITE_TABLETS_INTO, currentBatchId.get(), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilder.java index 231329cb26276..7d6967db7ec42 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilder.java @@ -147,7 +147,13 @@ private List> writeTabletsToTsFiles() // Try making the tsfile size as large as possible while (!device2TabletsLinkedList.isEmpty()) { if (Objects.isNull(fileWriter)) { - fileWriter = new TsFileWriter(createFile()); + final File file = createFile(); + try { + fileWriter = new TsFileWriter(file); + } catch (final IOException | RuntimeException e) { + FileUtils.deleteQuietly(file); + throw e; + } } try { tryBestToWriteTabletsIntoOneFile(device2TabletsLinkedList, device2Aligned); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilderV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilderV2.java index 07703695d16e3..4aedf542ea4ff 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilderV2.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilderV2.java @@ -30,6 +30,7 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.exception.write.WriteProcessException; +import org.apache.tsfile.external.commons.io.FileUtils; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.utils.DateUtils; import org.apache.tsfile.utils.Pair; @@ -120,10 +121,15 @@ public synchronized void close() { private List> writeTabletsToTsFiles() throws WriteProcessException { final IMemTable memTable = new PrimitiveMemTable(null, null); final List> sealedFiles = new ArrayList<>(); - try (final RestorableTsFileIOWriter writer = new RestorableTsFileIOWriter(createFile())) { - writeTabletsIntoOneFile(memTable, writer); - sealedFiles.add(new Pair<>(null, writer.getFile())); + File file = null; + try { + file = createFile(); + try (final RestorableTsFileIOWriter writer = new RestorableTsFileIOWriter(file)) { + writeTabletsIntoOneFile(memTable, writer); + sealedFiles.add(new Pair<>(null, writer.getFile())); + } } catch (final Exception e) { + FileUtils.deleteQuietly(file); LOGGER.warn( DataNodePipeMessages.BATCH_ID_FAILED_TO_WRITE_TABLETS_INTO, currentBatchId.get(), @@ -155,14 +161,16 @@ private void writeTabletsIntoOneFile( for (int j = 0; j < tablet.getSchemas().size(); ++j) { final IMeasurementSchema schema = measurementSchemas[j]; if (Objects.isNull(schema)) { - break; + continue; } if (Objects.equals(TSDataType.DATE, schema.getType()) && values[j] instanceof LocalDate[]) { final LocalDate[] dates = ((LocalDate[]) values[j]); final int[] dateValues = new int[dates.length]; for (int k = 0; k < Math.min(dates.length, tablet.getRowSize()); k++) { - dateValues[k] = DateUtils.parseDateExpressionToInt(dates[k]); + if (Objects.nonNull(dates[k])) { + dateValues[k] = DateUtils.parseDateExpressionToInt(dates[k]); + } } values[j] = dateValues; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index 63a6269b190fc..19686757e2905 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -202,6 +202,7 @@ import java.nio.file.Paths; import java.util.ArrayList; import java.util.Arrays; +import java.util.BitSet; import java.util.Collection; import java.util.Collections; import java.util.Comparator; @@ -833,7 +834,7 @@ protected void upgradeAndUpdateDeviceLastFlushTime( // checked above //noinspection OptionalGetWithoutIsPresent long endTime = resource.getEndTime(deviceId).get(); - endTimeMap.put(deviceId, endTime); + endTimeMap.merge(deviceId, endTime, Math::max); } } if (config.isEnableSeparateData()) { @@ -1344,14 +1345,36 @@ private void updateSplitInfo( rangeList.add(newRange); } + private Map[]> splitUnprocessedTabletRows( + final InsertTabletNode insertTabletNode, + final int start, + final List> deviceEndOffsetPairs, + final BitSet processedRows) { + final Map[]> splitInfo = new HashMap<>(); + int deviceStart = start; + for (final Pair deviceEndOffsetPair : deviceEndOffsetPairs) { + final int deviceEnd = deviceEndOffsetPair.getRight(); + int rangeStart = processedRows.nextClearBit(deviceStart); + while (rangeStart < deviceEnd) { + final int nextProcessed = processedRows.nextSetBit(rangeStart); + final int rangeEnd = + nextProcessed < 0 || nextProcessed > deviceEnd ? deviceEnd : nextProcessed; + split(insertTabletNode, rangeStart, rangeEnd, splitInfo); + rangeStart = processedRows.nextClearBit(rangeEnd); + } + deviceStart = deviceEnd; + } + return splitInfo; + } + private boolean doInsert( InsertTabletNode insertTabletNode, Map[]> splitMap, TSStatus[] results, long[] infoForMetrics, - boolean markLastFragmentOnFinalWrite) + boolean markLastFragmentOnFinalWrite, + TabletInsertionProgress progress) throws DataTypeInconsistentException { - boolean noFailure = true; int remainingFragmentCount = 0; if (markLastFragmentOnFinalWrite) { for (Entry[]> entry : splitMap.entrySet()) { @@ -1371,36 +1394,38 @@ private boolean doInsert( if (sequenceRangeList != null) { insertTabletNode.setLastFragment( markLastFragmentOnFinalWrite && remainingFragmentCount == 1); - noFailure = + final boolean fragmentSucceeded = insertTabletToTsFileProcessor( - insertTabletNode, - sequenceRangeList, - true, - results, - timePartitionId, - noFailure, - infoForMetrics) - && noFailure; + insertTabletNode, + sequenceRangeList, + true, + results, + timePartitionId, + progress.noFailure, + infoForMetrics); + markInsertTabletRangesProcessed(sequenceRangeList, progress.processedRows); + progress.noFailure = fragmentSucceeded && progress.noFailure; remainingFragmentCount--; } List unSequenceRangeList = rangeLists[0]; if (unSequenceRangeList != null) { insertTabletNode.setLastFragment( markLastFragmentOnFinalWrite && remainingFragmentCount == 1); - noFailure = + final boolean fragmentSucceeded = insertTabletToTsFileProcessor( - insertTabletNode, - unSequenceRangeList, - false, - results, - timePartitionId, - noFailure, - infoForMetrics) - && noFailure; + insertTabletNode, + unSequenceRangeList, + false, + results, + timePartitionId, + progress.noFailure, + infoForMetrics); + markInsertTabletRangesProcessed(unSequenceRangeList, progress.processedRows); + progress.noFailure = fragmentSucceeded && progress.noFailure; remainingFragmentCount--; } } - return noFailure; + return progress.noFailure; } /** @@ -1457,34 +1482,54 @@ private boolean splitAndInsert( List> deviceEndOffsetPairs, boolean markLastFragmentOnFinalWrite) { final int initialStart = start; + final TabletInsertionProgress progress = + new TabletInsertionProgress(insertTabletNode.getRowCount()); try { - Map[]> splitInfo = new HashMap<>(); - for (Pair deviceEndOffsetPair : deviceEndOffsetPairs) { - int end = deviceEndOffsetPair.getRight(); - split(insertTabletNode, start, end, splitInfo); - start = end; - } + final Map[]> splitInfo = + splitUnprocessedTabletRows( + insertTabletNode, initialStart, deviceEndOffsetPairs, progress.processedRows); return doInsert( - insertTabletNode, splitInfo, results, infoForMetrics, markLastFragmentOnFinalWrite); + insertTabletNode, + splitInfo, + results, + infoForMetrics, + markLastFragmentOnFinalWrite, + progress); } catch (DataTypeInconsistentException e) { // the exception will trigger a flush, which requires the flush time to be recalculated - start = initialStart; - Map[]> splitInfo = new HashMap<>(); - for (Pair deviceEndOffsetPair : deviceEndOffsetPairs) { - int end = deviceEndOffsetPair.getRight(); - split(insertTabletNode, start, end, splitInfo); - start = end; - } + final Map[]> splitInfo = + splitUnprocessedTabletRows( + insertTabletNode, initialStart, deviceEndOffsetPairs, progress.processedRows); try { return doInsert( - insertTabletNode, splitInfo, results, infoForMetrics, markLastFragmentOnFinalWrite); + insertTabletNode, + splitInfo, + results, + infoForMetrics, + markLastFragmentOnFinalWrite, + progress); } catch (DataTypeInconsistentException ex) { logger.error(StorageEngineMessages.DATA_INCONSISTENT_NOT_TRIGGER_TWICE, ex); + progress.noFailure = false; + for (int i = initialStart; i < insertTabletNode.getRowCount(); i++) { + if (!progress.processedRows.get(i)) { + results[i] = RpcUtils.getStatus(ex.getErrorCode(), ex.getMessage()); + } + } return false; } } } + private static final class TabletInsertionProgress { + private final BitSet processedRows; + private boolean noFailure = true; + + private TabletInsertionProgress(final int rowCount) { + processedRows = new BitSet(rowCount); + } + } + private boolean executeInsertTablet( InsertTabletNode insertTabletNode, TSStatus[] results, @@ -1604,6 +1649,8 @@ private boolean insertTabletToTsFileProcessor( throw e; } catch (WriteProcessRejectException e) { logger.warn(StorageEngineMessages.INSERT_TO_TSFILE_PROCESSOR_REJECTED, e.getMessage()); + markInsertTabletRangesFailed( + rangeList, results, RpcUtils.getStatus(e.getErrorCode(), e.getMessage())); return false; } catch (WriteProcessException e) { logger.error(StorageEngineMessages.INSERT_TO_TSFILE_PROCESSOR_ERROR, e); @@ -1617,6 +1664,13 @@ private boolean insertTabletToTsFileProcessor( return true; } + private void markInsertTabletRangesProcessed( + final List rangeList, final BitSet processedRows) { + for (final int[] rangePair : rangeList) { + processedRows.set(rangePair[0], rangePair[1]); + } + } + private void markInsertTabletRangesFailed( final List rangeList, final TSStatus[] results, final TSStatus failureStatus) { for (int[] rangePair : rangeList) { @@ -1826,10 +1880,11 @@ private List insertToTsFileProcessors( InsertRowsNode subInsertRowsNode = entry.getValue(); subInsertRowsNode.setLastFragment(--remainingFragments == 0); try { - List insertedProcessors = + InsertRowsExecutionResult executionResult = insertRowsWithTypeConsistencyCheck(entry.getKey(), subInsertRowsNode, infoForMetrics); - executedInsertRowNodeList.addAll(subInsertRowsNode.getInsertRowNodeList()); - for (TsFileProcessor tsFileProcessor : insertedProcessors) { + executedInsertRowNodeList.addAll(executionResult.insertedRows); + insertRowsNode.getResults().putAll(executionResult.failedResults); + for (TsFileProcessor tsFileProcessor : executionResult.insertedProcessors) { // check memtable size and may asyncTryToFlush the work memtable if (tsFileProcessor.shouldFlush()) { fileFlushPolicy.apply(this, tsFileProcessor, tsFileProcessor.isSequence()); @@ -1842,14 +1897,16 @@ private List insertToTsFileProcessors( return executedInsertRowNodeList; } - private List insertRowsWithTypeConsistencyCheck( + private InsertRowsExecutionResult insertRowsWithTypeConsistencyCheck( TsFileProcessor tsFileProcessor, InsertRowsNode subInsertRowsNode, long[] infoForMetrics) throws WriteProcessException { try { // register TableSchema (and maybe more) for table insertion registerToTsFile(subInsertRowsNode, tsFileProcessor); tsFileProcessor.insertRows(subInsertRowsNode, infoForMetrics); - return Collections.singletonList(tsFileProcessor); + InsertRowsExecutionResult executionResult = new InsertRowsExecutionResult(); + executionResult.recordSuccess(tsFileProcessor, subInsertRowsNode); + return executionResult; } catch (DataTypeInconsistentException e) { InsertRowNode firstRow = subInsertRowsNode.getInsertRowNodeList().get(0); long timePartitionId = TimePartitionUtils.getTimePartitionId(firstRow.getTime()); @@ -1926,11 +1983,11 @@ private void flushWorkingProcessorsForTimePartition(final long timePartitionId) } } - private List retryInsertRowsAfterFlush( + private InsertRowsExecutionResult retryInsertRowsAfterFlush( final InsertRowsNode subInsertRowsNode, final long timePartitionId, - final long[] infoForMetrics) - throws WriteProcessException { + final long[] infoForMetrics) { + final InsertRowsExecutionResult executionResult = new InsertRowsExecutionResult(); final Map retriedProcessorMap = new HashMap<>(); for (int i = 0; i < subInsertRowsNode.getInsertRowNodeList().size(); i++) { final InsertRowNode insertRowNode = subInsertRowsNode.getInsertRowNodeList().get(i); @@ -1938,9 +1995,14 @@ private List retryInsertRowsAfterFlush( config.isEnableSeparateData() && insertRowNode.getTime() > lastFlushTimeMap.getFlushedTime(timePartitionId, insertRowNode.getDeviceID()); - final TsFileProcessor retriedProcessor = - getOrCreateTsFileProcessor(timePartitionId, isSequence); final int insertRowNodeIndex = subInsertRowsNode.getInsertRowNodeIndexList().get(i); + final TsFileProcessor retriedProcessor; + try { + retriedProcessor = getOrCreateTsFileProcessor(timePartitionId, isSequence); + } catch (WriteProcessException e) { + executionResult.recordFailure(insertRowNodeIndex, e); + continue; + } retriedProcessorMap.compute( retriedProcessor, (k, v) -> { @@ -1952,18 +2014,47 @@ private List retryInsertRowsAfterFlush( }); } - final List insertedProcessors = new ArrayList<>(retriedProcessorMap.size()); int remainingRetriedFragments = retriedProcessorMap.size(); for (Entry retriedEntry : retriedProcessorMap.entrySet()) { final TsFileProcessor retriedProcessor = retriedEntry.getKey(); final InsertRowsNode retriedInsertRowsNode = retriedEntry.getValue(); retriedInsertRowsNode.setLastFragment( subInsertRowsNode.isLastFragment() && --remainingRetriedFragments == 0); - registerToTsFile(retriedInsertRowsNode, retriedProcessor); - retriedProcessor.insertRows(retriedInsertRowsNode, infoForMetrics); - insertedProcessors.add(retriedProcessor); + try { + registerToTsFile(retriedInsertRowsNode, retriedProcessor); + retriedProcessor.insertRows(retriedInsertRowsNode, infoForMetrics); + executionResult.recordSuccess(retriedProcessor, retriedInsertRowsNode); + } catch (WriteProcessException e) { + executionResult.recordFailure(retriedInsertRowsNode, e); + } + } + return executionResult; + } + + private static final class InsertRowsExecutionResult { + private final List insertedProcessors = new ArrayList<>(); + private final List insertedRows = new ArrayList<>(); + private final Map failedResults = new HashMap<>(); + + private void recordSuccess( + final TsFileProcessor tsFileProcessor, final InsertRowsNode insertRowsNode) { + insertedProcessors.add(tsFileProcessor); + insertedRows.addAll(insertRowsNode.getInsertRowNodeList()); + } + + private void recordFailure( + final InsertRowsNode insertRowsNode, final WriteProcessException exception) { + final TSStatus failureStatus = + RpcUtils.getStatus(exception.getErrorCode(), exception.getMessage()); + for (final int failedIndex : insertRowsNode.getInsertRowNodeIndexList()) { + failedResults.put(failedIndex, failureStatus); + } + } + + private void recordFailure(final int failedIndex, final WriteProcessException exception) { + failedResults.put( + failedIndex, RpcUtils.getStatus(exception.getErrorCode(), exception.getMessage())); } - return insertedProcessors; } private void recordInsertRowsFailure( @@ -4814,10 +4905,11 @@ public void insert(InsertRowsOfOneDeviceNode insertRowsOfOneDeviceNode) InsertRowsNode subInsertRowsNode = entry.getValue(); subInsertRowsNode.setLastFragment(--remainingFragments == 0); try { - List insertedProcessors = + InsertRowsExecutionResult executionResult = insertRowsWithTypeConsistencyCheck(entry.getKey(), subInsertRowsNode, infoForMetrics); - executedInsertRowNodeList.addAll(subInsertRowsNode.getInsertRowNodeList()); - for (TsFileProcessor tsFileProcessor : insertedProcessors) { + executedInsertRowNodeList.addAll(executionResult.insertedRows); + insertRowsOfOneDeviceNode.getResults().putAll(executionResult.failedResults); + for (TsFileProcessor tsFileProcessor : executionResult.insertedProcessors) { // check memtable size and may asyncTryToFlush the work memtable if (tsFileProcessor.shouldFlush()) { fileFlushPolicy.apply(this, tsFileProcessor, tsFileProcessor.isSequence()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/HashLastFlushTimeMap.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/HashLastFlushTimeMap.java index f29c158cc0dfd..2452cfd011125 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/HashLastFlushTimeMap.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/HashLastFlushTimeMap.java @@ -152,6 +152,7 @@ public long getFlushedTime(long timePartitionId, IDeviceID deviceId) { @Override public void clearFlushedTime() { partitionLatestFlushedTime.clear(); + memCostForEachPartition.clear(); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index 6dc5ecb3e2b28..2b2b4a14c9cb4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -398,7 +398,13 @@ public void insertRows(InsertRowsNode insertRowsNode, long[] infoForMetrics) } } long[] alignedMemIncrements = checkAlignedMemCostAndAddToTspInfoForRows(alignedList); - long[] nonAlignedMemIncrements = checkMemCostAndAddToTspInfoForRows(nonAlignedList); + final long[] nonAlignedMemIncrements; + try { + nonAlignedMemIncrements = checkMemCostAndAddToTspInfoForRows(nonAlignedList); + } catch (final WriteProcessException e) { + rollbackMemoryInfoIfNeeded(alignedMemIncrements); + throw e; + } memIncrements = new long[3]; for (int i = 0; i < 3; i++) { memIncrements[i] = alignedMemIncrements[i] + nonAlignedMemIncrements[i]; @@ -495,25 +501,29 @@ private long[] scheduleMemoryBlock( throws WriteProcessException { long memControlStartTime = System.nanoTime(); long[] totalMemIncrements = new long[NUM_MEM_TO_ESTIMATE]; - for (int[] range : rangeList) { - int start = range[0]; - int end = range[1]; - try { + try { + for (int[] range : rangeList) { + int start = range[0]; + int end = range[1]; long[] memIncrements = checkMemCost(insertTabletNode, start, end, results); for (int i = 0; i < memIncrements.length; i++) { totalMemIncrements[i] += memIncrements[i]; } - } catch (WriteProcessException e) { - for (int i = start; i < end; i++) { - results[i] = RpcUtils.getStatus(TSStatusCode.WRITE_PROCESS_REJECT, e.getMessage()); + } + return totalMemIncrements; + } catch (final WriteProcessException e) { + rollbackMemoryInfoIfNeeded(totalMemIncrements); + final TSStatus failureStatus = RpcUtils.getStatus(e.getErrorCode(), e.getMessage()); + for (final int[] range : rangeList) { + for (int i = range[0]; i < range[1]; i++) { + results[i] = failureStatus; } - throw new WriteProcessException(e); } + throw e; + } finally { + // recordScheduleMemoryBlockCost + infoForMetrics[1] += System.nanoTime() - memControlStartTime; } - // recordScheduleMemoryBlockCost - infoForMetrics[1] += System.nanoTime() - memControlStartTime; - - return totalMemIncrements; } private long[] checkMemCost( @@ -545,16 +555,22 @@ private long[] checkAlignedMemCost( for (Pair iDeviceIDIntegerPair : deviceEndPosList) { int splitEnd = iDeviceIDIntegerPair.getRight(); IDeviceID deviceID = iDeviceIDIntegerPair.getLeft(); - long[] splitMemIncrements = - checkAlignedMemCostAndAddToTspForTablet( - deviceID, - insertTabletNode.getMeasurements(), - insertTabletNode.getDataTypes(), - insertTabletNode.getColumns(), - insertTabletNode.getColumnCategories(), - splitStart, - splitEnd, - results); + final long[] splitMemIncrements; + try { + splitMemIncrements = + checkAlignedMemCostAndAddToTspForTablet( + deviceID, + insertTabletNode.getMeasurements(), + insertTabletNode.getDataTypes(), + insertTabletNode.getColumns(), + insertTabletNode.getColumnCategories(), + splitStart, + splitEnd, + results); + } catch (final WriteProcessException e) { + rollbackMemoryInfoIfNeeded(memIncrements); + throw e; + } for (int i = 0; i < NUM_MEM_TO_ESTIMATE; i++) { memIncrements[i] += splitMemIncrements[i]; } @@ -628,7 +644,8 @@ public void insertTablet( tsFileResource); int pointInserted = 0; - for (int[] rangePair : rangeList) { + for (int rangeIndex = 0; rangeIndex < rangeList.size(); rangeIndex++) { + final int[] rangePair = rangeList.get(rangeIndex); int start = rangePair[0]; int end = rangePair[1]; try { @@ -639,11 +656,17 @@ public void insertTablet( } else { pointInserted += workMemTable.insertTablet(insertTabletNode, start, end); } - } catch (WriteProcessException e) { - for (int i = start; i < end; i++) { - results[i] = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage()); + } catch (final WriteProcessException e) { + final TSStatus failureStatus = RpcUtils.getStatus(e.getErrorCode(), e.getMessage()); + for (int failedRangeIndex = rangeIndex; + failedRangeIndex < rangeList.size(); + failedRangeIndex++) { + final int[] failedRange = rangeList.get(failedRangeIndex); + for (int i = failedRange[0]; i < failedRange[1]; i++) { + results[i] = failureStatus; + } } - throw new WriteProcessException(e); + throw e; } for (int i = start; i < end; i++) { if (results[i] == null @@ -1219,6 +1242,15 @@ private void rollbackMemoryInfo(long[] memIncrements) { workMemTable.releaseTextDataSize(textDataIncrement); } + private void rollbackMemoryInfoIfNeeded(final long[] memIncrements) { + for (final long memIncrement : memIncrements) { + if (memIncrement != 0) { + rollbackMemoryInfo(memIncrements); + return; + } + } + } + /** * Delete data which belongs to the timeseries `deviceId.measurementId` and the timestamp of which * <= 'timestamp' in the deletion.
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java index 0e96c893fb766..a425dd1221079 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java @@ -31,6 +31,7 @@ import org.apache.iotdb.commons.consensus.ConfigRegionId; import org.apache.iotdb.commons.consensus.ConsensusGroupId; import org.apache.iotdb.commons.subscription.config.SubscriptionConfig; +import org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper; import org.apache.iotdb.confignode.rpc.thrift.TGetCommitProgressReq; import org.apache.iotdb.confignode.rpc.thrift.TGetCommitProgressResp; import org.apache.iotdb.db.conf.IoTDBDescriptor; @@ -127,7 +128,7 @@ public class ConsensusSubscriptionCommitManager { IoTDBThreadPoolFactory.newSingleThreadExecutor( ThreadName.SUBSCRIPTION_CONSENSUS_PROGRESS_BROADCASTER.getName()); - /** Key: "consumerGroupId##topicName##regionId" -> progress tracking state */ + /** Key: versioned, encoded (consumerGroupId, topicName, regionId) -> progress tracking state. */ private final Map commitStates = new ConcurrentHashMap<>(); @@ -414,13 +415,26 @@ public void persistAll() { public Map collectAllRegionProgress(final int dataNodeId) { recoverAllTopicStatesIfNeeded(); final Map result = new ConcurrentHashMap<>(); - final String suffix = KEY_SEPARATOR + dataNodeId; for (final Map.Entry entry : commitStates.entrySet()) { + final CommitStateKey stateKey = commitStateKeys.get(entry.getKey()); + if (Objects.isNull(stateKey)) { + continue; + } final RegionProgress regionProgress = entry.getValue().getCommittedRegionProgress(); final ByteBuffer serialized = serializeRegionProgress(regionProgress); if (Objects.nonNull(serialized)) { - result.put(entry.getKey() + suffix, serialized); + result.put( + CommitProgressKeeper.generateKey( + stateKey.consumerGroupId, stateKey.topicName, stateKey.regionIdStr, dataNodeId), + serialized); + if (CommitProgressKeeper.isLegacyKeyUnambiguous( + stateKey.consumerGroupId, stateKey.topicName, stateKey.regionIdStr)) { + result.put( + CommitProgressKeeper.generateLegacyKey( + stateKey.consumerGroupId, stateKey.topicName, stateKey.regionIdStr, dataNodeId), + serialized.asReadOnlyBuffer()); + } } } return result; @@ -578,7 +592,7 @@ public void receiveProgressBroadcast( // ======================== Helper Methods ======================== - // Kept as the in-memory and ConfigNode sync key separator for the existing progress protocol. + // Used for encoded topic file keys and writer-scoped broadcast throttle keys. private static final String KEY_SEPARATOR = "##"; private String generateKey( @@ -588,7 +602,7 @@ private String generateKey( private String generateKey( final String consumerGroupId, final String topicName, final String regionIdStr) { - return consumerGroupId + KEY_SEPARATOR + topicName + KEY_SEPARATOR + regionIdStr; + return CommitProgressKeeper.generateRegionKey(consumerGroupId, topicName, regionIdStr); } private CommitStateKey getCommitStateKey( @@ -1356,6 +1370,7 @@ public void updateFromBroadcast(final WriterId writerId, final WriterProgress wr final ProgressKey current = new ProgressKey(broadcastWriterId, currentWriterProgress); if (broadcastKey.compareTo(current) > 0) { committedWriterPositions.put(broadcastWriterId, broadcastKey.toWriterProgress()); + progress.incrementPersistenceThrottleCounter(); syncPersistedProgress(); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java index 6bb40f70d1bb3..f05c1e467bad8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatch.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.subscription.event.batch; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventTsFileBatch; import org.apache.iotdb.db.subscription.agent.SubscriptionAgent; @@ -32,6 +33,7 @@ import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent; import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionCommitContext; +import org.apache.tsfile.external.commons.io.FileUtils; import org.apache.tsfile.utils.Pair; import org.apache.tsfile.write.record.Tablet; import org.slf4j.Logger; @@ -49,6 +51,7 @@ public class SubscriptionPipeTsFileEventBatch extends SubscriptionPipeEventBatch LoggerFactory.getLogger(SubscriptionPipeTsFileEventBatch.class); private final PipeTabletEventTsFileBatch batch; + private final List> sealedFilePairs = new ArrayList<>(); public SubscriptionPipeTsFileEventBatch( final int regionId, @@ -61,6 +64,17 @@ public SubscriptionPipeTsFileEventBatch( maxDelayInMs, maxBatchSizeInBytes, this::pruneTableModelTablet); } + @TestOnly + SubscriptionPipeTsFileEventBatch( + final int regionId, + final SubscriptionPrefetchingTsFileQueue prefetchingQueue, + final int maxDelayInMs, + final long maxBatchSizeInBytes, + final PipeTabletEventTsFileBatch batch) { + super(regionId, prefetchingQueue, maxDelayInMs, maxBatchSizeInBytes); + this.batch = batch; + } + @Override public synchronized void ack() { batch.decreaseEventsReferenceCount(this.getClass().getName(), true); @@ -68,9 +82,19 @@ public synchronized void ack() { @Override public synchronized void cleanUp(final boolean force) { - // close batch, it includes clearing the reference count of events - batch.close(); - enrichedEvents.clear(); + try { + // close batch, it includes clearing the reference count of events + batch.close(); + } finally { + for (final Pair sealedFilePair : sealedFilePairs) { + final File sealedFile = sealedFilePair.right; + if (sealedFile.exists() && !FileUtils.deleteQuietly(sealedFile)) { + LOGGER.warn(DataNodePipeMessages.FAILED_TO_DELETE_BATCH_FILE_THIS_FILE, sealedFilePair); + } + } + sealedFilePairs.clear(); + enrichedEvents.clear(); + } } /////////////////////////////// utility /////////////////////////////// @@ -112,6 +136,7 @@ protected List generateSubscriptionEvents() throws Exception enrichedEvents.clear(); return Collections.emptyList(); } + sealedFilePairs.addAll(dbTsFilePairs); final AtomicInteger ackReferenceCount = new AtomicInteger(dbTsFilePairs.size()); final AtomicInteger cleanReferenceCount = new AtomicInteger(dbTsFilePairs.size()); for (final Pair pair : dbTsFilePairs) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandlerCleanupTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandlerCleanupTest.java new file mode 100644 index 0000000000000..aca597f0463ff --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandlerCleanupTest.java @@ -0,0 +1,86 @@ +/* + * 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.pipe.sink.protocol.thrift.async.handler; + +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; + +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +import java.io.File; +import java.nio.file.Files; +import java.util.Collections; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +public class PipeTransferTsFileHandlerCleanupTest { + + @Test + public void testCloseDeletesBatchFile() throws Exception { + final File file = Files.createTempFile("pipe-transfer-batch", ".tsfile").toFile(); + final EnrichedEvent event = Mockito.mock(EnrichedEvent.class); + + createHandler(file, event).close(); + + Assert.assertFalse(file.exists()); + } + + @Test + public void testNullClientDeletesBatchFile() throws Exception { + final File file = Files.createTempFile("pipe-transfer-null-client", ".tsfile").toFile(); + final EnrichedEvent event = Mockito.mock(EnrichedEvent.class); + final PipeTransferTsFileHandler handler = createHandler(file, event); + + handler.transfer(null, null); + + Assert.assertFalse(file.exists()); + } + + @Test + public void testCloseKeepsSourceTsFile() throws Exception { + final File file = Files.createTempFile("pipe-transfer-source", ".tsfile").toFile(); + final PipeTsFileInsertionEvent event = Mockito.mock(PipeTsFileInsertionEvent.class); + try { + createHandler(file, event).close(); + Assert.assertTrue(file.exists()); + } finally { + if (file.exists()) { + Assert.assertTrue(file.delete()); + } + } + } + + private PipeTransferTsFileHandler createHandler(final File file, final EnrichedEvent event) + throws Exception { + return new PipeTransferTsFileHandler( + Mockito.mock(IoTDBDataRegionAsyncSink.class), + Collections.emptyMap(), + Collections.singletonList(event), + new AtomicInteger(1), + new AtomicBoolean(false), + file, + null, + false, + null); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTsFileBuilderV2Test.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTsFileBuilderV2Test.java new file mode 100644 index 0000000000000..6e6ab5e1c1f13 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTsFileBuilderV2Test.java @@ -0,0 +1,264 @@ +/* + * 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.pipe.sink.util.builder; + +import org.apache.iotdb.db.conf.IoTDBDescriptor; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.external.commons.io.FileUtils; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.utils.Pair; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +public class PipeTsFileBuilderV2Test { + + private static File temporaryPipeReceiverDir; + private static String[] originalPipeReceiverFileDirs; + + @BeforeClass + public static void setUpClass() throws Exception { + originalPipeReceiverFileDirs = + IoTDBDescriptor.getInstance().getConfig().getPipeReceiverFileDirs(); + temporaryPipeReceiverDir = Files.createTempDirectory("pipe-table-builder-v2-test").toFile(); + IoTDBDescriptor.getInstance() + .getConfig() + .setPipeReceiverFileDirs(new String[] {temporaryPipeReceiverDir.getAbsolutePath()}); + } + + @AfterClass + public static void tearDownClass() throws Exception { + IoTDBDescriptor.getInstance().getConfig().setPipeReceiverFileDirs(originalPipeReceiverFileDirs); + FileUtils.deleteDirectory(temporaryPipeReceiverDir); + } + + @Test + public void testFallbackBuilderConvertsBufferedTablets() throws Exception { + final PipeTableModelTsFileBuilderV2 builder = + new PipeTableModelTsFileBuilderV2(new AtomicLong(1), new AtomicLong(0)) { + @Override + protected File createFile() throws IOException { + throw new IOException(); + } + }; + final List> fallbackFiles = new ArrayList<>(); + try { + final Tablet tablet = + new Tablet( + "table", + Arrays.asList("tag", "field"), + Arrays.asList(TSDataType.STRING, TSDataType.INT64), + Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD), + 1); + tablet.addTimestamp(0, 1); + tablet.addValue(0, 0, "device"); + tablet.addValue(0, 1, 1L); + + builder.bufferTableModelTablet("database", tablet); + fallbackFiles.addAll(builder.convertTabletToTsFileWithDBInfo()); + + Assert.assertEquals(1, fallbackFiles.size()); + Assert.assertEquals("database", fallbackFiles.get(0).left); + Assert.assertTrue(fallbackFiles.get(0).right.isFile()); + Assert.assertTrue(fallbackFiles.get(0).right.length() > 0); + } finally { + builder.close(); + fallbackFiles.forEach(file -> FileUtils.deleteQuietly(file.right)); + } + } + + @Test + public void testFallbackDeletesPrimaryTableModelFiles() throws Exception { + final AtomicInteger fileIndex = new AtomicInteger(); + final List primaryFiles = new ArrayList<>(); + final PipeTableModelTsFileBuilderV2 builder = + new PipeTableModelTsFileBuilderV2(new AtomicLong(2), new AtomicLong(0)) { + @Override + protected File createFile() throws IOException { + final int index = fileIndex.incrementAndGet(); + final File file = + new File(temporaryPipeReceiverDir, "primary-table-" + index + ".tsfile"); + primaryFiles.add(file); + if (index == 2 && !file.mkdir()) { + throw new IOException("Failed to create invalid primary file path"); + } + return file; + } + }; + final List> fallbackFiles = new ArrayList<>(); + try { + builder.bufferTableModelTablet("database1", createTableModelTablet("table1")); + builder.bufferTableModelTablet("database2", createTableModelTablet("table2")); + fallbackFiles.addAll(builder.convertTabletToTsFileWithDBInfo()); + + Assert.assertEquals(2, fallbackFiles.size()); + Assert.assertEquals(2, primaryFiles.size()); + primaryFiles.forEach(file -> Assert.assertFalse(file.exists())); + } finally { + builder.close(); + fallbackFiles.forEach(file -> FileUtils.deleteQuietly(file.right)); + primaryFiles.forEach(FileUtils::deleteQuietly); + } + } + + @Test + public void testClassicTableModelBuilderDeletesEarlierDatabaseFilesOnFailure() throws Exception { + final AtomicInteger fileIndex = new AtomicInteger(); + final List createdFiles = new ArrayList<>(); + final PipeTableModelTsFileBuilder builder = + new PipeTableModelTsFileBuilder(new AtomicLong(5), new AtomicLong(0)) { + @Override + protected File createFile() throws IOException { + final int index = fileIndex.incrementAndGet(); + final File file = + new File(temporaryPipeReceiverDir, "classic-table-" + index + ".tsfile"); + createdFiles.add(file); + if (index == 2) { + Assert.assertTrue(createdFiles.get(0).isFile()); + if (!file.mkdir()) { + throw new IOException("Failed to create invalid classic file path"); + } + } + return file; + } + }; + try { + builder.bufferTableModelTablet("database1", createTableModelTablet("table1")); + builder.bufferTableModelTablet("database2", createTableModelTablet("table2")); + + try { + builder.convertTabletToTsFileWithDBInfo(); + Assert.fail("Expected the second database conversion to fail"); + } catch (final Exception expected) { + // expected + } + + Assert.assertEquals(2, createdFiles.size()); + createdFiles.forEach(file -> Assert.assertFalse(file.exists())); + } finally { + builder.close(); + createdFiles.forEach(FileUtils::deleteQuietly); + } + } + + @Test + public void testTreeModelBuilderPreservesColumnsAfterNullSchema() throws Exception { + final PipeTreeModelTsFileBuilderV2 builder = + new PipeTreeModelTsFileBuilderV2(new AtomicLong(3), new AtomicLong(0)); + final List> files = new ArrayList<>(); + try { + final List schemas = + Arrays.asList( + new MeasurementSchema("s0", TSDataType.INT64, TSEncoding.PLAIN), + new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN), + new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.PLAIN)); + final Tablet tablet = + new Tablet( + "root.database.device", + schemas, + new long[] {1L}, + new Object[] {new long[] {1L}, new long[] {2L}, new long[] {3L}}, + null, + 1); + tablet.getSchemas().set(1, null); + + builder.bufferTreeModelTablet(tablet, false); + files.addAll(builder.convertTabletToTsFileWithDBInfo()); + + Assert.assertEquals(1, files.size()); + try (final TsFileSequenceReader reader = + new TsFileSequenceReader(files.get(0).right.getAbsolutePath())) { + final IDeviceID deviceID = IDeviceID.Factory.DEFAULT_FACTORY.create("root.database.device"); + Assert.assertNotNull(reader.readTimeseriesMetadata(deviceID, "s0", true)); + Assert.assertNotNull(reader.readTimeseriesMetadata(deviceID, "s2", true)); + } + } finally { + builder.close(); + files.forEach(file -> FileUtils.deleteQuietly(file.right)); + } + } + + @Test + public void testFallbackDeletesPrimaryTreeModelFile() throws Exception { + final File primaryFile = new File(temporaryPipeReceiverDir, "primary-tree.tsfile"); + final PipeTreeModelTsFileBuilderV2 builder = + new PipeTreeModelTsFileBuilderV2(new AtomicLong(4), new AtomicLong(0)) { + @Override + protected File createFile() throws IOException { + if (!primaryFile.mkdir()) { + throw new IOException("Failed to create invalid primary file path"); + } + return primaryFile; + } + }; + final List> fallbackFiles = new ArrayList<>(); + try { + final Tablet tablet = + new Tablet( + "root.database.device", + Arrays.asList( + new MeasurementSchema("s0", TSDataType.INT64, TSEncoding.PLAIN)), + 1); + tablet.addTimestamp(0, 1L); + tablet.addValue(0, 0, 1L); + + builder.bufferTreeModelTablet(tablet, false); + fallbackFiles.addAll(builder.convertTabletToTsFileWithDBInfo()); + + Assert.assertEquals(1, fallbackFiles.size()); + Assert.assertFalse(primaryFile.exists()); + } finally { + builder.close(); + fallbackFiles.forEach(file -> FileUtils.deleteQuietly(file.right)); + FileUtils.deleteQuietly(primaryFile); + } + } + + private static Tablet createTableModelTablet(final String tableName) { + final Tablet tablet = + new Tablet( + tableName, + Arrays.asList("tag", "field"), + Arrays.asList(TSDataType.STRING, TSDataType.INT64), + Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD), + 1); + tablet.addTimestamp(0, 1L); + tablet.addValue(0, 0, "device"); + tablet.addValue(0, 1, 1L); + return tablet; + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java index f2487eed6c948..1000ad31cc4c0 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java @@ -38,6 +38,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.BatchProcessException; import org.apache.iotdb.db.exception.DataRegionException; +import org.apache.iotdb.db.exception.DataTypeInconsistentException; import org.apache.iotdb.db.exception.TsFileProcessorException; import org.apache.iotdb.db.exception.WriteProcessException; import org.apache.iotdb.db.exception.WriteProcessRejectException; @@ -46,6 +47,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; @@ -108,9 +110,11 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; import static org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils.genInsertRowNode; import static org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils.genInsertTabletNode; @@ -1151,6 +1155,91 @@ public void testInsertRowsMarkAllFailedRowsForSameProcessor() throws Exception { } } + @Test + public void testInsertRowsTypeRetryKeepsSuccessfulFragments() throws Exception { + assertInsertRowsTypeRetryKeepsSuccessfulFragments(false); + } + + @Test + public void testInsertRowsOfOneDeviceTypeRetryKeepsSuccessfulFragments() throws Exception { + assertInsertRowsTypeRetryKeepsSuccessfulFragments(true); + } + + private void assertInsertRowsTypeRetryKeepsSuccessfulFragments(final boolean oneDevice) + throws Exception { + final boolean originalEnableSeparateData = config.isEnableSeparateData(); + config.setEnableSeparateData(true); + final String devicePath = oneDevice ? "root.retry_rows_one_device" : "root.retry_rows"; + final HookedDataRegion dataRegion1 = new HookedDataRegion(systemDir, devicePath); + try { + final TSRecord existingRecord = new TSRecord(devicePath, 5); + existingRecord.addTuple( + DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(5))); + dataRegion1.insert(buildInsertRowNodeByTSRecord(existingRecord)); + final TsFileProcessor initialProcessor = + dataRegion1.getWorkSequenceTsFileProcessors().iterator().next(); + + final TsFileProcessor successfulRetryProcessor = Mockito.mock(TsFileProcessor.class); + final TsFileProcessor failedRetryProcessor = Mockito.mock(TsFileProcessor.class); + Mockito.when(successfulRetryProcessor.shouldFlush()).thenReturn(false); + Mockito.doThrow(new WriteProcessException("mock retry fragment failure")) + .when(failedRetryProcessor) + .insertRows(any(InsertRowsNode.class), any(long[].class)); + + final AtomicInteger processorLookupCount = new AtomicInteger(); + dataRegion1.setTsFileProcessorSupplier( + (timePartitionId, sequence) -> { + if (processorLookupCount.getAndIncrement() < 2) { + return initialProcessor; + } + return sequence ? failedRetryProcessor : successfulRetryProcessor; + }); + + final List indexList = Arrays.asList(0, 1); + final List rows = new ArrayList<>(); + for (long time : new long[] {3, 7}) { + final TSRecord record = new TSRecord(devicePath, time); + record.addTuple( + DataPoint.getDataPoint(TSDataType.INT64, measurementId, String.valueOf(time))); + rows.add(buildInsertRowNodeByTSRecord(record)); + } + + if (oneDevice) { + final InsertRowsOfOneDeviceNode insertRowsNode = + new InsertRowsOfOneDeviceNode(new PlanNodeId(""), indexList, rows); + insertRowsNode.setTargetPath(new PartialPath(devicePath)); + try { + dataRegion1.insert(insertRowsNode); + Assert.fail("Expected BatchProcessException"); + } catch (BatchProcessException e) { + assertOnlySecondRetryFragmentFailed(insertRowsNode.getResults()); + } + } else { + final InsertRowsNode insertRowsNode = + new InsertRowsNode(new PlanNodeId(""), indexList, rows); + try { + dataRegion1.insert(insertRowsNode); + Assert.fail("Expected BatchProcessException"); + } catch (BatchProcessException e) { + assertOnlySecondRetryFragmentFailed(insertRowsNode.getResults()); + } + } + + Mockito.verify(successfulRetryProcessor) + .insertRows(any(InsertRowsNode.class), any(long[].class)); + Mockito.verify(failedRetryProcessor).insertRows(any(InsertRowsNode.class), any(long[].class)); + } finally { + dataRegion1.syncDeleteDataFiles(); + config.setEnableSeparateData(originalEnableSeparateData); + } + } + + private void assertOnlySecondRetryFragmentFailed(final Map results) { + Assert.assertEquals(1, results.size()); + Assert.assertFalse(results.containsKey(0)); + Assert.assertEquals(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode(), results.get(1).getCode()); + } + @Test public void testInsertRowsLastCacheSkipsFailedRows() throws Exception { final boolean originalLastCacheEnable = COMMON_CONFIG.isLastCacheEnable(); @@ -1279,6 +1368,107 @@ public void testInsertTabletLastCacheSkipsFailedRows() throws Exception { } } + @Test + public void testInsertTabletRejectMarksAllRowsFailed() throws Exception { + final HookedDataRegion dataRegion1 = new HookedDataRegion(systemDir, "root.reject_tablet"); + final TsFileProcessor processor = Mockito.mock(TsFileProcessor.class); + Mockito.doThrow(new WriteProcessRejectException("mock tablet rejection")) + .when(processor) + .insertTablet( + any(InsertTabletNode.class), + anyList(), + any(TSStatus[].class), + anyBoolean(), + any(long[].class)); + dataRegion1.setTsFileProcessorSupplier((timePartitionId, sequence) -> processor); + + final InsertTabletNode insertTabletNode = + new InsertTabletNode( + new QueryId("test_write").genPlanNodeId(), + new PartialPath("root.reject_tablet"), + false, + new String[] {measurementId}, + new TSDataType[] {TSDataType.INT64}, + new MeasurementSchema[] { + new MeasurementSchema(measurementId, TSDataType.INT64, TSEncoding.PLAIN) + }, + new long[] {1L, 2L}, + null, + new Object[] {new long[] {1L, 2L}}, + 2); + + try { + dataRegion1.insertTablet(insertTabletNode); + Assert.fail("Expected BatchProcessException"); + } catch (final BatchProcessException e) { + for (final TSStatus status : e.getFailingStatus()) { + Assert.assertEquals(TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode(), status.getCode()); + } + } finally { + dataRegion1.syncDeleteDataFiles(); + } + } + + @Test + public void testInsertTabletTypeRetryDoesNotReplaySuccessfulFragments() throws Exception { + final HookedDataRegion dataRegion1 = new HookedDataRegion(systemDir, "root.retry_tablet"); + final TsFileProcessor firstProcessor = Mockito.mock(TsFileProcessor.class); + final TsFileProcessor secondProcessor = Mockito.mock(TsFileProcessor.class); + final AtomicInteger insertionCount = new AtomicInteger(); + final org.mockito.stubbing.Answer insertionAnswer = + invocation -> { + if (insertionCount.incrementAndGet() == 2) { + throw new DataTypeInconsistentException(TSDataType.INT32, TSDataType.INT64); + } + return null; + }; + Mockito.doAnswer(insertionAnswer) + .when(firstProcessor) + .insertTablet( + any(InsertTabletNode.class), + anyList(), + any(TSStatus[].class), + anyBoolean(), + any(long[].class)); + Mockito.doAnswer(insertionAnswer) + .when(secondProcessor) + .insertTablet( + any(InsertTabletNode.class), + anyList(), + any(TSStatus[].class), + anyBoolean(), + any(long[].class)); + + final long firstTime = 1L; + final long secondTime = firstTime + TimePartitionUtils.getTimePartitionInterval(); + final long firstPartitionId = TimePartitionUtils.getTimePartitionId(firstTime); + Assert.assertNotEquals(firstPartitionId, TimePartitionUtils.getTimePartitionId(secondTime)); + dataRegion1.setTsFileProcessorSupplier( + (timePartitionId, sequence) -> + timePartitionId == firstPartitionId ? firstProcessor : secondProcessor); + + final InsertTabletNode insertTabletNode = + new InsertTabletNode( + new QueryId("test_write").genPlanNodeId(), + new PartialPath("root.retry_tablet"), + false, + new String[] {measurementId}, + new TSDataType[] {TSDataType.INT64}, + new MeasurementSchema[] { + new MeasurementSchema(measurementId, TSDataType.INT64, TSEncoding.PLAIN) + }, + new long[] {firstTime, secondTime}, + null, + new Object[] {new long[] {1L, 2L}}, + 2); + try { + dataRegion1.insertTablet(insertTabletNode); + Assert.assertEquals(3, insertionCount.get()); + } finally { + dataRegion1.syncDeleteDataFiles(); + } + } + @Test public void testSmallReportProportionInsertRow() throws WriteProcessException, diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/LastFlushTimeMapTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/LastFlushTimeMapTest.java index d14d1f27a403d..7f58ccdd22831 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/LastFlushTimeMapTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/LastFlushTimeMapTest.java @@ -38,9 +38,13 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.mockito.Mockito; import java.io.File; import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.Optional; public class LastFlushTimeMapTest { @@ -139,6 +143,35 @@ record = new TSRecord("root.vehicle.d0", j); .getFlushedTime(0, IDeviceID.Factory.DEFAULT_FACTORY.create("root.vehicle.d1"))); } + @Test + public void testClearFlushedTimeClearsMemoryCost() { + HashLastFlushTimeMap lastFlushTimeMap = new HashLastFlushTimeMap(); + IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create("root.vehicle.d0"); + + lastFlushTimeMap.updateMultiDeviceFlushedTime(0, Collections.singletonMap(device, 100L)); + Assert.assertTrue(lastFlushTimeMap.getMemSize(0) > 0); + lastFlushTimeMap.clearFlushedTime(); + Assert.assertEquals(0, lastFlushTimeMap.getMemSize(0)); + } + + @Test + public void testUpgradeDeviceFlushTimeUsesMaximumAcrossResources() { + final IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create("root.vehicle.d0"); + final TsFileResource newerEndTimeResource = Mockito.mock(TsFileResource.class); + final TsFileResource olderEndTimeResource = Mockito.mock(TsFileResource.class); + Mockito.when(newerEndTimeResource.getDevices()).thenReturn(Collections.singleton(device)); + Mockito.when(newerEndTimeResource.getEndTime(device)).thenReturn(Optional.of(100L)); + Mockito.when(olderEndTimeResource.getDevices()).thenReturn(Collections.singleton(device)); + Mockito.when(olderEndTimeResource.getEndTime(device)).thenReturn(Optional.of(50L)); + + dataRegion.getLastFlushTimeMap().clearFlushedTime(); + dataRegion.getLastFlushTimeMap().checkAndCreateFlushedTimePartition(0, true); + dataRegion.upgradeAndUpdateDeviceLastFlushTime( + 0, Arrays.asList(newerEndTimeResource, olderEndTimeResource)); + + Assert.assertEquals(100L, dataRegion.getLastFlushTimeMap().getFlushedTime(0, device)); + } + @Test public void testRecoverLastFlushTimeMap() throws IOException, IllegalPathException, WriteProcessException, DataRegionException { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java index 159b81f73ad85..229188f8dea85 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.subscription.broker.consensus; import org.apache.iotdb.commons.consensus.DataRegionId; +import org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress; import org.apache.iotdb.rpc.subscription.payload.poll.WriterId; @@ -168,9 +169,25 @@ public void testDirectCommitWithoutOutstandingRespectsOutstandingGap() { assertEquals(new WriterProgress(103L, 3L), state.getCommittedWriterProgress()); } + @Test + public void testBroadcastAdvanceIncrementsPersistenceCounter() { + final String regionId = "3_3"; + final WriterId writerId = new WriterId(regionId, 8); + final SubscriptionConsensusProgress progress = + new SubscriptionConsensusProgress(new RegionProgress(new LinkedHashMap<>()), 7L); + final ConsensusSubscriptionCommitManager.ConsensusSubscriptionCommitState state = + new ConsensusSubscriptionCommitManager.ConsensusSubscriptionCommitState(regionId, progress); + + state.updateFromBroadcast(writerId, new WriterProgress(101L, 1L)); + assertEquals(8L, progress.getPersistenceThrottleCounter()); + + state.updateFromBroadcast(writerId, new WriterProgress(100L, 0L)); + assertEquals(8L, progress.getPersistenceThrottleCounter()); + } + @Test public void testBroadcastThrottleKeyIsPerWriter() { - final String baseKey = "cg##topic##1_1"; + final String baseKey = CommitProgressKeeper.generateRegionKey("cg", "topic", "1_1"); final WriterId writerA = new WriterId("1_1", 7); final WriterId writerB = new WriterId("1_1", 8); @@ -213,12 +230,23 @@ public void testPersistGroupedTopicProgressAndRecoverAllRegions() throws Excepti assertEquals( progress1, - RegionProgress.deserialize(collectedProgress.get("cg##topic##" + region1Id + "##11")) + RegionProgress.deserialize( + collectedProgress.get( + CommitProgressKeeper.generateKey("cg", "topic", region1Id, 11))) + .getWriterPositions() + .get(writer1)); + assertEquals( + progress1, + RegionProgress.deserialize( + collectedProgress.get( + CommitProgressKeeper.generateLegacyKey("cg", "topic", region1Id, 11))) .getWriterPositions() .get(writer1)); assertEquals( progress2, - RegionProgress.deserialize(collectedProgress.get("cg##topic##" + region2Id + "##11")) + RegionProgress.deserialize( + collectedProgress.get( + CommitProgressKeeper.generateKey("cg", "topic", region2Id, 11))) .getWriterPositions() .get(writer2)); } finally { @@ -258,7 +286,7 @@ public void testProgressFileNameEncodesForbiddenTopicComponents() throws Excepti progress, RegionProgress.deserialize( collectedProgress.get( - consumerGroupId + "##" + topicName + "##" + regionId + "##13")) + CommitProgressKeeper.generateKey(consumerGroupId, topicName, regionId, 13))) .getWriterPositions() .get(writerId)); } finally { @@ -286,7 +314,9 @@ public void testRecoverAllSkipsMalformedTopicProgressFile() throws Exception { assertEquals( progress, - RegionProgress.deserialize(collectedProgress.get("cg##goodTopic##" + regionId + "##12")) + RegionProgress.deserialize( + collectedProgress.get( + CommitProgressKeeper.generateKey("cg", "goodTopic", regionId, 12))) .getWriterPositions() .get(writerId)); } finally { @@ -294,6 +324,44 @@ public void testRecoverAllSkipsMalformedTopicProgressFile() throws Exception { } } + @Test + public void testCommitStatesDoNotCollideWhenIdentifiersContainSeparator() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("systemWithCollidingLegacyKeys"); + try { + final ConsensusSubscriptionCommitManager manager = newCommitManager(systemDir); + final DataRegionId region = new DataRegionId(10); + final String regionId = region.toString(); + final WriterId firstWriterId = new WriterId(regionId, 7); + final WriterId secondWriterId = new WriterId(regionId, 8); + final WriterProgress firstProgress = new WriterProgress(100L, 1L); + final WriterProgress secondProgress = new WriterProgress(200L, 2L); + + manager.receiveProgressBroadcast("a##b", "c", regionId, firstWriterId, firstProgress); + manager.receiveProgressBroadcast("a", "b##c", regionId, secondWriterId, secondProgress); + + final Map collectedProgress = + newCommitManager(systemDir).collectAllRegionProgress(14); + assertEquals(2, collectedProgress.size()); + assertEquals( + firstProgress, + RegionProgress.deserialize( + collectedProgress.get( + CommitProgressKeeper.generateKey("a##b", "c", regionId, 14))) + .getWriterPositions() + .get(firstWriterId)); + assertEquals( + secondProgress, + RegionProgress.deserialize( + collectedProgress.get( + CommitProgressKeeper.generateKey("a", "b##c", regionId, 14))) + .getWriterPositions() + .get(secondWriterId)); + } finally { + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + private static void writeMalformedTopicProgressFile(final File systemDir) throws Exception { final File progressDir = new File(systemDir, "subscription/consensus_progress"); final File malformedFile = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatchCleanupTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatchCleanupTest.java new file mode 100644 index 0000000000000..d4d3609bbdbbd --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTsFileEventBatchCleanupTest.java @@ -0,0 +1,78 @@ +/* + * 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.event.batch; + +import org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventTsFileBatch; +import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingTsFileQueue; +import org.apache.iotdb.db.subscription.event.SubscriptionEvent; +import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionCommitContext; + +import org.apache.tsfile.external.commons.io.FileUtils; +import org.apache.tsfile.utils.Pair; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +import java.io.File; +import java.nio.file.Files; +import java.util.Arrays; +import java.util.List; + +public class SubscriptionPipeTsFileEventBatchCleanupTest { + + @Test + public void testDeletesSealedFilesAfterAllSharedEventsAreCleaned() throws Exception { + final File temporaryDirectory = + Files.createTempDirectory("subscription-tsfile-cleanup").toFile(); + final File firstFile = new File(temporaryDirectory, "first.tsfile"); + final File secondFile = new File(temporaryDirectory, "second.tsfile"); + Assert.assertTrue(firstFile.createNewFile()); + Assert.assertTrue(secondFile.createNewFile()); + + final PipeTabletEventTsFileBatch innerBatch = Mockito.mock(PipeTabletEventTsFileBatch.class); + final SubscriptionPrefetchingTsFileQueue queue = + Mockito.mock(SubscriptionPrefetchingTsFileQueue.class); + Mockito.when(innerBatch.isEmpty()).thenReturn(false); + Mockito.when(innerBatch.sealTsFiles()) + .thenReturn(Arrays.asList(new Pair<>("db1", firstFile), new Pair<>("db2", secondFile))); + Mockito.when(queue.generateSubscriptionCommitContext()) + .thenReturn( + new SubscriptionCommitContext(1, 1, "topic", "group", 1), + new SubscriptionCommitContext(1, 1, "topic", "group", 2)); + + try { + final SubscriptionPipeTsFileEventBatch batch = + new SubscriptionPipeTsFileEventBatch(1, queue, 1, 1, innerBatch); + final List events = batch.generateSubscriptionEvents(); + + Assert.assertEquals(2, events.size()); + events.get(0).cleanUp(false); + Assert.assertTrue(firstFile.exists()); + Assert.assertTrue(secondFile.exists()); + + events.get(1).cleanUp(false); + Assert.assertFalse(firstFile.exists()); + Assert.assertFalse(secondFile.exists()); + Mockito.verify(innerBatch, Mockito.times(1)).close(); + } finally { + FileUtils.deleteDirectory(temporaryDirectory); + } + } +} diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java index afb977f85a204..4bece78f64135 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -902,6 +902,9 @@ private PipeMessages() {} public static final String EXCEPTION_UNEXPECTED_EOF_READING_REGION_PROGRESS_KEY_C1532EAE = "Unexpected EOF reading region progress key"; public static final String EXCEPTION_UNEXPECTED_EOF_READING_REGION_PROGRESS_VALUE_LENGTH_D95F9CE0 = "Unexpected EOF reading region progress value length"; public static final String EXCEPTION_UNEXPECTED_EOF_READING_REGION_PROGRESS_VALUE_A459C521 = "Unexpected EOF reading region progress value"; + public static final String EXCEPTION_INVALID_REGION_PROGRESS_ENTRY_COUNT_B43DED2F = "Invalid region progress entry count: %d"; + public static final String EXCEPTION_INVALID_REGION_PROGRESS_KEY_LENGTH_7C3A3C98 = "Invalid region progress key length: %d"; + public static final String EXCEPTION_INVALID_REGION_PROGRESS_VALUE_LENGTH_6192D17F = "Invalid region progress value length: %d"; public static final String EXCEPTION_FAILED_ADD_SUBSCRIPTION_CONSUMER_GROUP_META_CONSUMER_ARG_DOES_NOT_EF08EE87 = "Failed to add subscription to consumer group meta: consumer %s does not exist in consumer" + " group %s"; diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java index 87f1601a7544b..c0fceb9a8c0f3 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -877,6 +877,9 @@ private PipeMessages() {} public static final String EXCEPTION_UNEXPECTED_EOF_READING_REGION_PROGRESS_KEY_C1532EAE = "读取 region progress key 时遇到意外 EOF"; public static final String EXCEPTION_UNEXPECTED_EOF_READING_REGION_PROGRESS_VALUE_LENGTH_D95F9CE0 = "读取 region progress value 长度时遇到意外 EOF"; public static final String EXCEPTION_UNEXPECTED_EOF_READING_REGION_PROGRESS_VALUE_A459C521 = "读取 region progress value 时遇到意外 EOF"; + public static final String EXCEPTION_INVALID_REGION_PROGRESS_ENTRY_COUNT_B43DED2F = "region progress entry 数量无效：%d"; + public static final String EXCEPTION_INVALID_REGION_PROGRESS_KEY_LENGTH_7C3A3C98 = "region progress key 长度无效：%d"; + public static final String EXCEPTION_INVALID_REGION_PROGRESS_VALUE_LENGTH_6192D17F = "region progress value 长度无效：%d"; public static final String EXCEPTION_FAILED_ADD_SUBSCRIPTION_CONSUMER_GROUP_META_CONSUMER_ARG_DOES_NOT_EF08EE87 = "添加 subscription 到 consumer group meta 失败：consumer %s 不存在于 consumer group %s"; public static final String EXCEPTION_FAILED_REMOVE_SUBSCRIPTION_CONSUMER_GROUP_META_CONSUMER_ARG_DOES_NOT_75C319C3 = "从 consumer group meta 移除 subscription 失败：consumer %s 不存在于 consumer group %s"; public static final String EXCEPTION_PATH_PATTERN_ARG_NOT_VALID_SOURCE_ONLY_PREFIX_FULL_PATH_784778B8 = "路径模式 %s 对 source 无效。仅允许前缀或完整路径。"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/queue/ConcurrentIterableLinkedQueue.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/queue/ConcurrentIterableLinkedQueue.java index 3e3eb1bd9ef55..a73c6cc66eb9d 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/queue/ConcurrentIterableLinkedQueue.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/queue/ConcurrentIterableLinkedQueue.java @@ -138,6 +138,7 @@ public long tryRemoveBefore(long newFirstIndex) { if (firstNode == null) { firstNode = pilotNode; lastNode = pilotNode; + pilotNode.next = null; } // Update iterators if necessary @@ -209,9 +210,7 @@ public void setFirstIndex(final long firstIndex) { lock.writeLock().lock(); try { this.firstIndex = firstIndex; - if (tailIndex < firstIndex) { - tailIndex = firstIndex; - } + tailIndex = firstIndex; } finally { lock.writeLock().unlock(); } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/queue/listening/AbstractSerializableListeningQueue.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/queue/listening/AbstractSerializableListeningQueue.java index 4c5465edc8ac3..de43350d4ad6f 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/queue/listening/AbstractSerializableListeningQueue.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/queue/listening/AbstractSerializableListeningQueue.java @@ -139,7 +139,7 @@ public synchronized void deserializeFromFile(final File snapshotName) throws IOE return; } - queue.clear(); + close(); try (final FileInputStream inputStream = new FileInputStream(snapshotFile)) { isClosed.set(ReadWriteIOUtils.readBool(inputStream)); final QueueSerializerType type = diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java index 4dc79a8f86155..7da9f7e157c9f 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java @@ -21,12 +21,15 @@ import org.apache.iotdb.commons.i18n.PipeMessages; +import java.io.DataInputStream; import java.io.DataOutputStream; +import java.io.EOFException; import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.util.Base64; import java.util.HashMap; import java.util.Map; import java.util.Objects; @@ -35,23 +38,108 @@ public class CommitProgressKeeper { private static final String KEY_SEPARATOR = "##"; + private static final String KEY_COMPONENT_SEPARATOR = "."; + private static final String VERSIONED_KEY_PREFIX = "v2:"; private final Map regionProgressMap = new ConcurrentHashMap<>(); public CommitProgressKeeper() {} + public static String generateRegionKey( + final String consumerGroupId, final String topicName, final String regionId) { + return VERSIONED_KEY_PREFIX + + encodeKeyComponent(consumerGroupId) + + KEY_COMPONENT_SEPARATOR + + encodeKeyComponent(topicName) + + KEY_COMPONENT_SEPARATOR + + encodeKeyComponent(regionId); + } + + public static String generateRegionKeyPrefix( + final String consumerGroupId, final String topicName, final String regionId) { + return generateRegionKey(consumerGroupId, topicName, regionId) + KEY_SEPARATOR; + } + public static String generateKey( final String consumerGroupId, final String topicName, final String regionId, final int dataNodeId) { - return consumerGroupId - + KEY_SEPARATOR - + topicName - + KEY_SEPARATOR - + regionId - + KEY_SEPARATOR - + dataNodeId; + return generateRegionKeyPrefix(consumerGroupId, topicName, regionId) + dataNodeId; + } + + public static String generateLegacyRegionKeyPrefix( + final String consumerGroupId, final String topicName, final String regionId) { + return consumerGroupId + KEY_SEPARATOR + topicName + KEY_SEPARATOR + regionId + KEY_SEPARATOR; + } + + public static String generateLegacyKey( + final String consumerGroupId, + final String topicName, + final String regionId, + final int dataNodeId) { + return generateLegacyRegionKeyPrefix(consumerGroupId, topicName, regionId) + dataNodeId; + } + + public static boolean isLegacyKeyUnambiguous( + final String consumerGroupId, final String topicName, final String regionId) { + return !String.valueOf(consumerGroupId).contains(KEY_SEPARATOR) + && !String.valueOf(topicName).contains(KEY_SEPARATOR) + && !String.valueOf(regionId).contains(KEY_SEPARATOR); + } + + public static boolean isValidDataNodeProgressKey(final String key, final String keyPrefix) { + if (!key.startsWith(keyPrefix)) { + return false; + } + return isCanonicalDataNodeId(key.substring(keyPrefix.length())); + } + + public void removeTopicProgress(final String consumerGroupId, final String topicName) { + final String versionedTopicKeyPrefix = + VERSIONED_KEY_PREFIX + + encodeKeyComponent(consumerGroupId) + + KEY_COMPONENT_SEPARATOR + + encodeKeyComponent(topicName) + + KEY_COMPONENT_SEPARATOR; + final String legacyTopicKeyPrefix = + String.valueOf(consumerGroupId) + KEY_SEPARATOR + String.valueOf(topicName) + KEY_SEPARATOR; + final boolean legacyTopicKeyIsUnambiguous = + isLegacyKeyUnambiguous(consumerGroupId, topicName, ""); + regionProgressMap + .keySet() + .removeIf( + key -> + isValidRegionAndDataNodeProgressKey(key, versionedTopicKeyPrefix) + || (legacyTopicKeyIsUnambiguous + && isValidRegionAndDataNodeProgressKey(key, legacyTopicKeyPrefix))); + } + + private static boolean isValidRegionAndDataNodeProgressKey( + final String key, final String topicKeyPrefix) { + if (!key.startsWith(topicKeyPrefix)) { + return false; + } + final String suffix = key.substring(topicKeyPrefix.length()); + final int separatorIndex = suffix.indexOf(KEY_SEPARATOR); + return separatorIndex > 0 + && separatorIndex == suffix.lastIndexOf(KEY_SEPARATOR) + && isCanonicalDataNodeId(suffix.substring(separatorIndex + KEY_SEPARATOR.length())); + } + + private static boolean isCanonicalDataNodeId(final String value) { + try { + final int dataNodeId = Integer.parseInt(value); + return dataNodeId >= 0 && Integer.toString(dataNodeId).equals(value); + } catch (final NumberFormatException e) { + return false; + } + } + + private static String encodeKeyComponent(final String component) { + return Base64.getUrlEncoder() + .withoutPadding() + .encodeToString(String.valueOf(component).getBytes(StandardCharsets.UTF_8)); } public void updateRegionProgress(final String key, final ByteBuffer committedRegionProgress) { @@ -93,32 +181,49 @@ public void processTakeSnapshot(final FileOutputStream fileOutputStream) throws public void processLoadSnapshot(final FileInputStream fileInputStream) throws IOException { regionProgressMap.clear(); + final DataInputStream dataInputStream = new DataInputStream(fileInputStream); final byte[] sizeBytes = new byte[4]; - if (fileInputStream.read(sizeBytes) != 4) { + final int firstSizeByte = dataInputStream.read(); + if (firstSizeByte < 0) { return; } + sizeBytes[0] = (byte) firstSizeByte; + dataInputStream.readFully(sizeBytes, 1, sizeBytes.length - 1); final int regionSize = ByteBuffer.wrap(sizeBytes).getInt(); + validateLength( + regionSize, + dataInputStream.available() / (2 * Integer.BYTES), + PipeMessages.EXCEPTION_INVALID_REGION_PROGRESS_ENTRY_COUNT_B43DED2F); for (int i = 0; i < regionSize; i++) { final byte[] keyLenBytes = new byte[4]; - if (fileInputStream.read(keyLenBytes) != 4) { + if (!readFully(dataInputStream, keyLenBytes)) { throw new IOException( PipeMessages.EXCEPTION_UNEXPECTED_EOF_READING_REGION_PROGRESS_KEY_LENGTH_EBC10484); } final int keyLen = ByteBuffer.wrap(keyLenBytes).getInt(); + final int remainingEntryCount = regionSize - i - 1; + validateLength( + keyLen, + dataInputStream.available() - Integer.BYTES - remainingEntryCount * 2 * Integer.BYTES, + PipeMessages.EXCEPTION_INVALID_REGION_PROGRESS_KEY_LENGTH_7C3A3C98); final byte[] keyBytes = new byte[keyLen]; - if (fileInputStream.read(keyBytes) != keyLen) { + if (!readFully(dataInputStream, keyBytes)) { throw new IOException( PipeMessages.EXCEPTION_UNEXPECTED_EOF_READING_REGION_PROGRESS_KEY_C1532EAE); } final String key = new String(keyBytes, StandardCharsets.UTF_8); final byte[] valueLenBytes = new byte[4]; - if (fileInputStream.read(valueLenBytes) != 4) { + if (!readFully(dataInputStream, valueLenBytes)) { throw new IOException( PipeMessages.EXCEPTION_UNEXPECTED_EOF_READING_REGION_PROGRESS_VALUE_LENGTH_D95F9CE0); } final int valueLen = ByteBuffer.wrap(valueLenBytes).getInt(); + validateLength( + valueLen, + dataInputStream.available() - remainingEntryCount * 2 * Integer.BYTES, + PipeMessages.EXCEPTION_INVALID_REGION_PROGRESS_VALUE_LENGTH_6192D17F); final byte[] valueBytes = new byte[valueLen]; - if (fileInputStream.read(valueBytes) != valueLen) { + if (!readFully(dataInputStream, valueBytes)) { throw new IOException( PipeMessages.EXCEPTION_UNEXPECTED_EOF_READING_REGION_PROGRESS_VALUE_A459C521); } @@ -126,6 +231,23 @@ public void processLoadSnapshot(final FileInputStream fileInputStream) throws IO } } + private static void validateLength(final int value, final int maximum, final String message) + throws IOException { + if (value < 0 || value > maximum) { + throw new IOException(String.format(message, value)); + } + } + + private static boolean readFully(final DataInputStream stream, final byte[] bytes) + throws IOException { + try { + stream.readFully(bytes); + return true; + } catch (final EOFException ignored) { + return false; + } + } + public void serializeToStream(final DataOutputStream stream) throws IOException { serializeRegionProgressMapToStream(regionProgressMap, stream); } @@ -155,13 +277,26 @@ public static Map deserializeRegionProgressFromBuffer( return new HashMap<>(); } final int size = buffer.getInt(); - final Map result = new HashMap<>(size); + validateLengthArgument( + size, + buffer.remaining() / (2 * Integer.BYTES), + PipeMessages.EXCEPTION_INVALID_REGION_PROGRESS_ENTRY_COUNT_B43DED2F); + final Map result = new HashMap<>(); for (int i = 0; i < size; i++) { final int keyLen = buffer.getInt(); + final int remainingEntryCount = size - i - 1; + validateLengthArgument( + keyLen, + buffer.remaining() - Integer.BYTES - remainingEntryCount * 2 * Integer.BYTES, + PipeMessages.EXCEPTION_INVALID_REGION_PROGRESS_KEY_LENGTH_7C3A3C98); final byte[] keyBytes = new byte[keyLen]; buffer.get(keyBytes); final String key = new String(keyBytes, StandardCharsets.UTF_8); final int valueLen = buffer.getInt(); + validateLengthArgument( + valueLen, + buffer.remaining() - remainingEntryCount * 2 * Integer.BYTES, + PipeMessages.EXCEPTION_INVALID_REGION_PROGRESS_VALUE_LENGTH_6192D17F); final byte[] valueBytes = new byte[valueLen]; buffer.get(valueBytes); result.put(key, ByteBuffer.wrap(valueBytes).asReadOnlyBuffer()); @@ -169,6 +304,13 @@ public static Map deserializeRegionProgressFromBuffer( return result; } + private static void validateLengthArgument( + final int value, final int maximum, final String message) { + if (value < 0 || value > maximum) { + throw new IllegalArgumentException(String.format(message, value)); + } + } + private static ByteBuffer copyBuffer(final ByteBuffer buffer) { final ByteBuffer duplicate = buffer.asReadOnlyBuffer(); duplicate.rewind(); diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/ConcurrentIterableLinkedQueueTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/ConcurrentIterableLinkedQueueTest.java index 7c18e26c1ab32..f2b0898b418de 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/ConcurrentIterableLinkedQueueTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/ConcurrentIterableLinkedQueueTest.java @@ -481,4 +481,25 @@ public void testIteratorSeek() { Assert.assertTrue(itr.hasNext()); Assert.assertEquals(Integer.valueOf(5), itr.next()); } + + @Test(timeout = 60000) + public void testSetFirstIndexReplacesEmptyQueueIndexes() { + queue.add(1); + queue.add(2); + queue.clear(); + + queue.setFirstIndex(1); + + Assert.assertEquals(1, queue.getFirstIndex()); + Assert.assertEquals(1, queue.getTailIndex()); + queue.add(3); + Assert.assertEquals(2, queue.getTailIndex()); + + try (final ConcurrentIterableLinkedQueue.DynamicIterator itr = + queue.iterateFromEarliest()) { + Assert.assertTrue(itr.hasNext()); + Assert.assertEquals(Integer.valueOf(3), itr.next()); + Assert.assertFalse(itr.hasNext()); + } + } } diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/queue/listening/AbstractPipeListeningQueueTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/queue/listening/AbstractPipeListeningQueueTest.java new file mode 100644 index 0000000000000..10755b4178731 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/queue/listening/AbstractPipeListeningQueueTest.java @@ -0,0 +1,91 @@ +/* + * 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.commons.pipe.datastructure.queue.listening; + +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.commons.pipe.event.PipeSnapshotEvent; +import org.apache.iotdb.pipe.api.event.Event; + +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +import java.io.File; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Collections; +import java.util.List; + +public class AbstractPipeListeningQueueTest { + + @Test + public void testDeserializeReleasesExistingQueueAndSnapshotCache() throws Exception { + final Path temporaryDirectory = Files.createTempDirectory("pipe-listening-queue-snapshot"); + final File snapshotFile = temporaryDirectory.resolve("queue.snapshot").toFile(); + final TestPipeListeningQueue emptyQueue = new TestPipeListeningQueue(); + final TestPipeListeningQueue queue = new TestPipeListeningQueue(); + final EnrichedEvent queuedEvent = Mockito.mock(EnrichedEvent.class); + final PipeSnapshotEvent cachedSnapshot = Mockito.mock(PipeSnapshotEvent.class); + try { + Assert.assertTrue(emptyQueue.serializeToFile(snapshotFile)); + + queue.open(); + queue.addEvent(queuedEvent); + queue.addSnapshots(Collections.singletonList(cachedSnapshot)); + Assert.assertEquals(1, queue.getSize()); + Assert.assertEquals(1, queue.findAvailableSnapshots(false).getRight().size()); + + queue.deserializeFromFile(snapshotFile); + + Mockito.verify(queuedEvent) + .decreaseReferenceCount(AbstractPipeListeningQueue.class.getName(), false); + Mockito.verify(cachedSnapshot) + .decreaseReferenceCount(AbstractPipeListeningQueue.class.getName(), false); + Assert.assertEquals(0, queue.getSize()); + Assert.assertTrue(queue.findAvailableSnapshots(false).getRight().isEmpty()); + Assert.assertFalse(queue.isOpened()); + } finally { + Files.deleteIfExists(snapshotFile.toPath()); + Files.deleteIfExists(temporaryDirectory); + } + } + + private static final class TestPipeListeningQueue extends AbstractPipeListeningQueue { + + private void addEvent(final EnrichedEvent event) { + tryListen(event); + } + + private void addSnapshots(final List snapshots) { + tryListen(snapshots); + } + + @Override + protected ByteBuffer serializeToByteBuffer(final Event event) { + return ByteBuffer.allocate(0); + } + + @Override + protected Event deserializeFromByteBuffer(final ByteBuffer byteBuffer) { + return null; + } + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeperTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeperTest.java index 3b01328d804c1..2e76874f4d5e1 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeperTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeperTest.java @@ -29,6 +29,7 @@ import java.io.DataOutputStream; import java.io.FileInputStream; import java.io.FileOutputStream; +import java.io.IOException; import java.nio.ByteBuffer; import java.nio.file.Files; import java.nio.file.Path; @@ -36,10 +37,70 @@ import java.util.Map; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; public class CommitProgressKeeperTest { + @Test + public void testVersionedKeysAvoidLegacySeparatorCollision() { + final String firstKey = CommitProgressKeeper.generateKey("a##b", "c", "1_1", 3); + final String secondKey = CommitProgressKeeper.generateKey("a", "b##c", "1_1", 3); + + assertNotEquals(firstKey, secondKey); + assertEquals( + CommitProgressKeeper.generateLegacyKey("a##b", "c", "1_1", 3), + CommitProgressKeeper.generateLegacyKey("a", "b##c", "1_1", 3)); + assertTrue( + firstKey.startsWith(CommitProgressKeeper.generateRegionKeyPrefix("a##b", "c", "1_1"))); + } + + @Test + public void testValidatesProgressKeyGrammarAndLegacyEligibility() { + final String prefix = CommitProgressKeeper.generateRegionKeyPrefix("cg", "topic", "1_1"); + + assertTrue( + CommitProgressKeeper.isValidDataNodeProgressKey( + CommitProgressKeeper.generateKey("cg", "topic", "1_1", 3), prefix)); + assertFalse(CommitProgressKeeper.isValidDataNodeProgressKey(prefix + "3##4", prefix)); + assertFalse(CommitProgressKeeper.isValidDataNodeProgressKey(prefix + "03", prefix)); + assertTrue(CommitProgressKeeper.isLegacyKeyUnambiguous("cg", "topic", "1_1")); + assertFalse(CommitProgressKeeper.isLegacyKeyUnambiguous("cg##suffix", "topic", "1_1")); + } + + @Test + public void testRemoveTopicProgressRemovesVersionedAndLegacyKeys() { + final CommitProgressKeeper keeper = new CommitProgressKeeper(); + final ByteBuffer progress = ByteBuffer.wrap(new byte[] {1}); + final String versionedKey = CommitProgressKeeper.generateKey("cg", "topic", "1_1", 1); + final String legacyKey = CommitProgressKeeper.generateLegacyKey("cg", "topic", "1_2", 2); + final String delimiterKey = CommitProgressKeeper.generateKey("a##b", "c", "1_3", 3); + final String ambiguousLegacyKey = CommitProgressKeeper.generateLegacyKey("a##b", "c", "1_3", 3); + final String otherTopicKey = CommitProgressKeeper.generateKey("cg", "other", "1_1", 1); + final String masqueradingLegacyKey = + CommitProgressKeeper.generateLegacyKey( + CommitProgressKeeper.generateRegionKey("cg", "topic", ""), "legacyTopic", "1_4", 4); + keeper.updateRegionProgress(versionedKey, progress); + keeper.updateRegionProgress(legacyKey, progress); + keeper.updateRegionProgress(delimiterKey, progress); + keeper.updateRegionProgress(ambiguousLegacyKey, progress); + keeper.updateRegionProgress(otherTopicKey, progress); + keeper.updateRegionProgress(masqueradingLegacyKey, progress); + + keeper.removeTopicProgress("cg", "topic"); + keeper.removeTopicProgress("a##b", "c"); + + assertNull(keeper.getRegionProgress(versionedKey)); + assertNull(keeper.getRegionProgress(legacyKey)); + assertNull(keeper.getRegionProgress(delimiterKey)); + assertNotNull(keeper.getRegionProgress(ambiguousLegacyKey)); + assertNotNull(keeper.getRegionProgress(otherTopicKey)); + assertNotNull(keeper.getRegionProgress(masqueradingLegacyKey)); + } + @Test public void testUpdateAndReplaceAllUseDefensiveCopies() throws Exception { final CommitProgressKeeper keeper = new CommitProgressKeeper(); @@ -103,6 +164,135 @@ public void testSnapshotRoundTripPreservesRegionProgress() throws Exception { } } + @Test + public void testSnapshotLoadHandlesShortReads() throws Exception { + final CommitProgressKeeper keeper = new CommitProgressKeeper(); + final String key = CommitProgressKeeper.generateKey("cg", "topic", "1_1", 3); + final RegionProgress regionProgress = createRegionProgress("1_1", 7, 100L, 10L); + keeper.updateRegionProgress(key, serialize(regionProgress)); + + final Path snapshot = Files.createTempFile("commit-progress-keeper-short-read", ".snapshot"); + try { + try (FileOutputStream fos = new FileOutputStream(snapshot.toFile())) { + keeper.processTakeSnapshot(fos); + } + + final CommitProgressKeeper restored = new CommitProgressKeeper(); + try (FileInputStream fis = new OneByteAtATimeFileInputStream(snapshot)) { + restored.processLoadSnapshot(fis); + } + + assertEquals(regionProgress, RegionProgress.deserialize(restored.getRegionProgress(key))); + } finally { + Files.deleteIfExists(snapshot); + } + } + + @Test + public void testSnapshotLoadRejectsTruncatedSizeHeader() throws Exception { + final Path snapshot = + Files.createTempFile("commit-progress-keeper-truncated-size", ".snapshot"); + try { + Files.write(snapshot, new byte[0]); + try (FileInputStream fis = new FileInputStream(snapshot.toFile())) { + new CommitProgressKeeper().processLoadSnapshot(fis); + } + + Files.write(snapshot, new byte[] {0, 0, 0}); + try (FileInputStream fis = new FileInputStream(snapshot.toFile())) { + try { + new CommitProgressKeeper().processLoadSnapshot(fis); + org.junit.Assert.fail("Expected IOException for a truncated size header"); + } catch (final IOException expected) { + // expected + } + } + } finally { + Files.deleteIfExists(snapshot); + } + } + + @Test + public void testSnapshotLoadRejectsNegativeLengths() throws Exception { + final Path snapshot = Files.createTempFile("commit-progress-keeper-negative", ".snapshot"); + try { + try (DataOutputStream dos = new DataOutputStream(new FileOutputStream(snapshot.toFile()))) { + dos.writeInt(-1); + } + assertSnapshotLoadFails(snapshot); + + try (DataOutputStream dos = new DataOutputStream(new FileOutputStream(snapshot.toFile()))) { + dos.writeInt(1); + dos.writeInt(-1); + } + assertSnapshotLoadFails(snapshot); + + try (DataOutputStream dos = new DataOutputStream(new FileOutputStream(snapshot.toFile()))) { + dos.writeInt(1); + dos.writeInt(1); + dos.writeByte('k'); + dos.writeInt(-1); + } + assertSnapshotLoadFails(snapshot); + } finally { + Files.deleteIfExists(snapshot); + } + } + + @Test + public void testSnapshotLoadRejectsLengthsExceedingRemainingBytes() throws Exception { + final Path snapshot = Files.createTempFile("commit-progress-keeper-oversized", ".snapshot"); + try { + try (DataOutputStream dos = new DataOutputStream(new FileOutputStream(snapshot.toFile()))) { + dos.writeInt(Integer.MAX_VALUE); + } + assertSnapshotLoadFails(snapshot); + + try (DataOutputStream dos = new DataOutputStream(new FileOutputStream(snapshot.toFile()))) { + dos.writeInt(1); + dos.writeInt(Integer.MAX_VALUE); + } + assertSnapshotLoadFails(snapshot); + + try (DataOutputStream dos = new DataOutputStream(new FileOutputStream(snapshot.toFile()))) { + dos.writeInt(1); + dos.writeInt(0); + dos.writeInt(Integer.MAX_VALUE); + } + assertSnapshotLoadFails(snapshot); + } finally { + Files.deleteIfExists(snapshot); + } + } + + @Test + public void testRegionProgressMapDeserializationRejectsNegativeLengths() { + assertDeserializeRegionProgressFails(ByteBuffer.allocate(Integer.BYTES).putInt(-1).flip()); + assertDeserializeRegionProgressFails( + ByteBuffer.allocate(2 * Integer.BYTES).putInt(1).putInt(-1).flip()); + assertDeserializeRegionProgressFails( + ByteBuffer.allocate(3 * Integer.BYTES + 1) + .putInt(1) + .putInt(1) + .put((byte) 'k') + .putInt(-1) + .flip()); + } + + @Test + public void testRegionProgressMapDeserializationRejectsLengthsExceedingRemainingBytes() { + assertDeserializeRegionProgressFails( + ByteBuffer.allocate(Integer.BYTES).putInt(Integer.MAX_VALUE).flip()); + assertDeserializeRegionProgressFails( + ByteBuffer.allocate(2 * Integer.BYTES).putInt(1).putInt(Integer.MAX_VALUE).flip()); + assertDeserializeRegionProgressFails( + ByteBuffer.allocate(3 * Integer.BYTES) + .putInt(1) + .putInt(0) + .putInt(Integer.MAX_VALUE) + .flip()); + } + @Test public void testRegionProgressMapSerializationRoundTrip() throws Exception { final String firstKey = CommitProgressKeeper.generateKey("cg", "topicA", "1_1", 3); @@ -125,6 +315,26 @@ public void testRegionProgressMapSerializationRoundTrip() throws Exception { assertEquals(secondProgress, RegionProgress.deserialize(restored.get(secondKey))); } + private static void assertSnapshotLoadFails(final Path snapshot) throws Exception { + try (FileInputStream fis = new FileInputStream(snapshot.toFile())) { + try { + new CommitProgressKeeper().processLoadSnapshot(fis); + org.junit.Assert.fail("Expected IOException for invalid snapshot lengths"); + } catch (final IOException expected) { + // expected + } + } + } + + private static void assertDeserializeRegionProgressFails(final ByteBuffer buffer) { + try { + CommitProgressKeeper.deserializeRegionProgressFromBuffer(buffer); + org.junit.Assert.fail("Expected IllegalArgumentException for invalid buffer lengths"); + } catch (final IllegalArgumentException expected) { + // expected + } + } + private static RegionProgress createRegionProgress( final String regionId, final int nodeId, final long physicalTime, final long localSeq) { return createRegionProgress( @@ -160,4 +370,21 @@ private static ByteBuffer serialize(final RegionProgress regionProgress) throws return ByteBuffer.wrap(baos.toByteArray()).asReadOnlyBuffer(); } } + + private static class OneByteAtATimeFileInputStream extends FileInputStream { + + private OneByteAtATimeFileInputStream(final Path path) throws IOException { + super(path.toFile()); + } + + @Override + public int read(final byte[] bytes) throws IOException { + return read(bytes, 0, bytes.length); + } + + @Override + public int read(final byte[] bytes, final int offset, final int length) throws IOException { + return super.read(bytes, offset, Math.min(length, 1)); + } + } }