This is an automated email from the ASF dual-hosted git repository.
sumitagrawl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new d2ae71ea087 HDDS-15065. Replace Ratis Snapshot Trigger with DB Flush
for Periodic Flush Operations (#10100)
d2ae71ea087 is described below
commit d2ae71ea087c29d86724e8487e0a55137048b0fe
Author: Priyesh Karatha <[email protected]>
AuthorDate: Mon Jun 1 10:02:33 2026 +0530
HDDS-15065. Replace Ratis Snapshot Trigger with DB Flush for Periodic Flush
Operations (#10100)
---
.../org/apache/hadoop/hdds/scm/ScmConfigKeys.java | 9 -
.../common/src/main/resources/ozone-default.xml | 6 -
.../org/apache/hadoop/hdds/scm/ha/RatisUtil.java | 3 -
.../hdds/scm/ha/SCMHADBTransactionBuffer.java | 7 +
.../hdds/scm/ha/SCMHADBTransactionBufferImpl.java | 72 ++++--
.../hdds/scm/ha/SCMHADBTransactionBufferStub.java | 13 +
.../hadoop/hdds/scm/ha/SCMHAManagerImpl.java | 3 +-
.../scm/ha/SCMHATransactionBufferMonitorTask.java | 26 +-
.../apache/hadoop/hdds/scm/ha/SCMStateMachine.java | 41 +--
.../ha/TestSCMHATransactionBufferMonitorTask.java | 275 +++++++++++++++++++++
.../AbstractTestStorageDistributionEndpoint.java | 2 -
.../scm/TestDatanodeSCMNodesReconfiguration.java | 1 -
.../hdds/scm/TestStorageContainerManagerHA.java | 1 -
13 files changed, 382 insertions(+), 77 deletions(-)
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java
index e0b28548e8e..2ff192d4663 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java
@@ -584,15 +584,6 @@ public final class ScmConfigKeys {
public static final long OZONE_SCM_HA_RATIS_SNAPSHOT_THRESHOLD_DEFAULT =
1000L;
- /**
- * the config will transfer value to ratis config
- * raft.server.snapshot.creation.gap, used by ratis to take snapshot
- * when manual trigger using api.
- */
- public static final String OZONE_SCM_HA_RATIS_SNAPSHOT_GAP
- = "ozone.scm.ha.ratis.server.snapshot.creation.gap";
- public static final long OZONE_SCM_HA_RATIS_SNAPSHOT_GAP_DEFAULT =
- 1024L;
public static final String OZONE_SCM_HA_RATIS_SNAPSHOT_DIR =
"ozone.scm.ha.ratis.snapshot.dir";
diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml
b/hadoop-hdds/common/src/main/resources/ozone-default.xml
index 3089132ca0b..3d1d94735e1 100644
--- a/hadoop-hdds/common/src/main/resources/ozone-default.xml
+++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml
@@ -4201,12 +4201,6 @@
topology cluster tree from SCM.
</description>
</property>
- <property>
- <name>ozone.scm.ha.ratis.server.snapshot.creation.gap</name>
- <value>1024</value>
- <tag>SCM, OZONE</tag>
- <description>Raft snapshot gap index after which snapshot can be
taken.</description>
- </property>
<property>
<name>ozone.scm.ha.dbtransactionbuffer.flush.interval</name>
<value>60s</value>
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/RatisUtil.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/RatisUtil.java
index f2900e38f40..eb5c429861b 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/RatisUtil.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/RatisUtil.java
@@ -231,9 +231,6 @@ private static void setRaftSnapshotProperties(
Snapshot.setAutoTriggerThreshold(properties,
ozoneConf.getLong(ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_THRESHOLD,
ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_THRESHOLD_DEFAULT));
- Snapshot.setCreationGap(properties,
- ozoneConf.getLong(ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_GAP,
- ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_GAP_DEFAULT));
}
public static void checkRatisException(IOException e, String port,
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBuffer.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBuffer.java
index 7acf03424d8..f1d376d9de7 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBuffer.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBuffer.java
@@ -44,7 +44,14 @@ public interface SCMHADBTransactionBuffer
void flush() throws RocksDatabaseException, CodecException;
+ void flushIfNeeded(long snapshotWaitTime)
+ throws RocksDatabaseException, CodecException;
+
boolean shouldFlush(long snapshotWaitTime);
void init() throws RocksDatabaseException, CodecException;
+
+ void beginApplyingTransaction();
+
+ void endApplyingTransaction();
}
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBufferImpl.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBufferImpl.java
index 4b1243fd53d..4baf97a56eb 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBufferImpl.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBufferImpl.java
@@ -20,6 +20,7 @@
import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
import com.google.common.base.Preconditions;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -53,6 +54,7 @@ public class SCMHADBTransactionBufferImpl implements
SCMHADBTransactionBuffer {
private final AtomicReference<SnapshotInfo> latestSnapshot
= new AtomicReference<>();
private final AtomicLong txFlushPending = new AtomicLong(0);
+ private final AtomicInteger applyingTransactions = new AtomicInteger(0);
private long lastSnapshotTimeMs = 0;
private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
@@ -124,30 +126,64 @@ public AtomicReference<SnapshotInfo>
getLatestSnapshotRef() {
public void flush() throws RocksDatabaseException, CodecException {
rwLock.writeLock().lock();
try {
- // write latest trx info into trx table in the same batch
- Table<String, TransactionInfo> transactionInfoTable
- = metadataStore.getTransactionInfoTable();
- transactionInfoTable.putWithBatch(currentBatchOperation,
- TRANSACTION_INFO_KEY, latestTrxInfo);
+ flushUnderWriteLock();
+ } finally {
+ rwLock.writeLock().unlock();
+ }
+ }
- metadataStore.getStore().commitBatchOperation(currentBatchOperation);
- currentBatchOperation.close();
- this.latestSnapshot.set(latestTrxInfo.toSnapshotInfo());
- // reset batch operation
- currentBatchOperation = metadataStore.getStore().initBatchOperation();
-
- DeletedBlockLog deletedBlockLog = scm.getScmBlockManager()
- .getDeletedBlockLog();
- Preconditions.checkArgument(
- deletedBlockLog instanceof DeletedBlockLogImpl);
- ((DeletedBlockLogImpl) deletedBlockLog).onFlush();
+ @Override
+ public void flushIfNeeded(long snapshotWaitTime)
+ throws RocksDatabaseException, CodecException {
+ rwLock.writeLock().lock();
+ try {
+ if (applyingTransactions.get() > 0) {
+ return;
+ }
+ long timeDiff = scm.getSystemClock().millis() - lastSnapshotTimeMs;
+ if (txFlushPending.get() > 0 && timeDiff > snapshotWaitTime) {
+ LOG.debug("Running TransactionFlushTask");
+ flushUnderWriteLock();
+ }
} finally {
- txFlushPending.set(0);
- lastSnapshotTimeMs = scm.getSystemClock().millis();
rwLock.writeLock().unlock();
}
}
+ private void flushUnderWriteLock()
+ throws RocksDatabaseException, CodecException {
+ // write latest trx info into trx table in the same batch
+ Table<String, TransactionInfo> transactionInfoTable
+ = metadataStore.getTransactionInfoTable();
+ transactionInfoTable.putWithBatch(currentBatchOperation,
+ TRANSACTION_INFO_KEY, latestTrxInfo);
+
+ metadataStore.getStore().commitBatchOperation(currentBatchOperation);
+ currentBatchOperation.close();
+ this.latestSnapshot.set(latestTrxInfo.toSnapshotInfo());
+ // reset batch operation
+ currentBatchOperation = metadataStore.getStore().initBatchOperation();
+
+ DeletedBlockLog deletedBlockLog = scm.getScmBlockManager()
+ .getDeletedBlockLog();
+ Preconditions.checkArgument(
+ deletedBlockLog instanceof DeletedBlockLogImpl);
+ ((DeletedBlockLogImpl) deletedBlockLog).onFlush();
+
+ txFlushPending.set(0);
+ lastSnapshotTimeMs = scm.getSystemClock().millis();
+ }
+
+ @Override
+ public void beginApplyingTransaction() {
+ applyingTransactions.incrementAndGet();
+ }
+
+ @Override
+ public void endApplyingTransaction() {
+ applyingTransactions.decrementAndGet();
+ }
+
@Override
public void init() throws RocksDatabaseException, CodecException {
metadataStore = scm.getScmMetadataStore();
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBufferStub.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBufferStub.java
index 2e7b3fdb0dd..c8347dc7738 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBufferStub.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBufferStub.java
@@ -101,6 +101,19 @@ public AtomicReference<SnapshotInfo>
getLatestSnapshotRef() {
return null;
}
+ @Override
+ public void flushIfNeeded(long snapshotWaitTime) throws
RocksDatabaseException {
+ flush();
+ }
+
+ @Override
+ public void beginApplyingTransaction() {
+ }
+
+ @Override
+ public void endApplyingTransaction() {
+ }
+
@Override
public void flush() throws RocksDatabaseException {
rwLock.writeLock().lock();
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManagerImpl.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManagerImpl.java
index 0cc600160e1..01f1fee0d36 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManagerImpl.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManagerImpl.java
@@ -142,8 +142,7 @@ private void createStartTransactionBufferMonitor() {
OZONE_SCM_HA_DBTRANSACTIONBUFFER_FLUSH_INTERVAL_DEFAULT,
TimeUnit.MILLISECONDS);
SCMHATransactionBufferMonitorTask monitorTask
- = new SCMHATransactionBufferMonitorTask(
- transactionBuffer, ratisServer, interval);
+ = new SCMHATransactionBufferMonitorTask(transactionBuffer, interval);
trxBufferMonitorService =
new BackgroundSCMService.Builder().setClock(scm.getSystemClock())
.setScmContext(scm.getScmContext())
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHATransactionBufferMonitorTask.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHATransactionBufferMonitorTask.java
index 85faedae1c3..74b8a059ae3 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHATransactionBufferMonitorTask.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHATransactionBufferMonitorTask.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.hdds.scm.ha;
import java.io.IOException;
-import org.apache.ratis.statemachine.SnapshotInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,7 +28,6 @@
public class SCMHATransactionBufferMonitorTask implements Runnable {
private static final Logger LOG =
LoggerFactory.getLogger(SCMHATransactionBufferMonitorTask.class);
- private final SCMRatisServer server;
private final SCMHADBTransactionBuffer transactionBuffer;
private final long flushInterval;
@@ -37,31 +35,17 @@ public class SCMHATransactionBufferMonitorTask implements
Runnable {
* SCMService related variables.
*/
public SCMHATransactionBufferMonitorTask(
- SCMHADBTransactionBuffer transactionBuffer,
- SCMRatisServer server, long flushInterval) {
+ SCMHADBTransactionBuffer transactionBuffer, long flushInterval) {
this.flushInterval = flushInterval;
this.transactionBuffer = transactionBuffer;
- this.server = server;
}
@Override
public void run() {
- if (transactionBuffer.shouldFlush(flushInterval)) {
- LOG.debug("Running TransactionFlushTask");
- // set latest snapshot to null for force snapshot
- // the value will be reset again when snapshot is taken
- final SnapshotInfo lastSnapshot = transactionBuffer
- .getLatestSnapshotRef().getAndSet(null);
- try {
- server.triggerSnapshot();
- } catch (IOException e) {
- LOG.error("Snapshot request is failed", e);
- } finally {
- // under failure case, if unable to take snapshot, its value
- // is reset to previous known value
- transactionBuffer.getLatestSnapshotRef().compareAndSet(
- null, lastSnapshot);
- }
+ try {
+ transactionBuffer.flushIfNeeded(flushInterval);
+ } catch (IOException e) {
+ LOG.error("TransactionFlushTask is failed", e);
}
}
}
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java
index f43ce2c3dc0..a8a7d6068f0 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java
@@ -160,6 +160,7 @@ public CompletableFuture<Message> applyTransaction(
final TransactionContext trx) {
final CompletableFuture<Message> applyTransactionFuture =
new CompletableFuture<>();
+ transactionBuffer.beginApplyingTransaction();
try {
final SCMRatisRequest request = SCMRatisRequest.decode(
Message.valueOf(trx.getStateMachineLogEntry().getLogData()));
@@ -192,6 +193,8 @@ public CompletableFuture<Message> applyTransaction(
} catch (Exception ex) {
applyTransactionFuture.completeExceptionally(ex);
ExitUtils.terminate(1, ex.getMessage(), ex, StateMachine.LOG);
+ } finally {
+ transactionBuffer.endApplyingTransaction();
}
return applyTransactionFuture;
}
@@ -363,23 +366,28 @@ public long takeSnapshot() throws IOException {
return lastAppliedIndex;
}
- long startTime = Time.monotonicNow();
+ transactionBuffer.beginApplyingTransaction();
+ try {
+ long startTime = Time.monotonicNow();
- TransactionInfo latestTrxInfo = transactionBuffer.getLatestTrxInfo();
- final TransactionInfo lastAppliedTrxInfo =
TransactionInfo.valueOf(lastTermIndex);
+ TransactionInfo latestTrxInfo = transactionBuffer.getLatestTrxInfo();
+ final TransactionInfo lastAppliedTrxInfo =
TransactionInfo.valueOf(lastTermIndex);
- if (latestTrxInfo.compareTo(lastAppliedTrxInfo) < 0) {
- transactionBuffer.updateLatestTrxInfo(lastAppliedTrxInfo);
- transactionBuffer.setLatestSnapshot(lastAppliedTrxInfo.toSnapshotInfo());
- } else {
- lastAppliedIndex = latestTrxInfo.getTransactionIndex();
- }
+ if (latestTrxInfo.compareTo(lastAppliedTrxInfo) < 0) {
+ transactionBuffer.updateLatestTrxInfo(lastAppliedTrxInfo);
+
transactionBuffer.setLatestSnapshot(lastAppliedTrxInfo.toSnapshotInfo());
+ } else {
+ lastAppliedIndex = latestTrxInfo.getTransactionIndex();
+ }
- transactionBuffer.flush();
+ transactionBuffer.flush();
- LOG.info("Current Snapshot Index {}, takeSnapshot took {} ms",
- lastAppliedIndex, Time.monotonicNow() - startTime);
- return lastAppliedIndex;
+ LOG.info("Current Snapshot Index {}, takeSnapshot took {} ms",
+ lastAppliedIndex, Time.monotonicNow() - startTime);
+ return lastAppliedIndex;
+ } finally {
+ transactionBuffer.endApplyingTransaction();
+ }
}
@Override
@@ -399,7 +407,12 @@ public void notifyTermIndexUpdated(long term, long index) {
}
if (transactionBuffer != null) {
- transactionBuffer.updateLatestTrxInfo(TransactionInfo.valueOf(term,
index));
+ transactionBuffer.beginApplyingTransaction();
+ try {
+ transactionBuffer.updateLatestTrxInfo(TransactionInfo.valueOf(term,
index));
+ } finally {
+ transactionBuffer.endApplyingTransaction();
+ }
}
if (currentLeaderTerm.get() == term) {
diff --git
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestSCMHATransactionBufferMonitorTask.java
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestSCMHATransactionBufferMonitorTask.java
new file mode 100644
index 00000000000..5f312c6c4dd
--- /dev/null
+++
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestSCMHATransactionBufferMonitorTask.java
@@ -0,0 +1,275 @@
+/*
+ * 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.hadoop.hdds.scm.ha;
+
+import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.protobuf.ByteString;
+import java.io.File;
+import java.time.Clock;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.scm.block.BlockManager;
+import org.apache.hadoop.hdds.scm.block.DeletedBlockLogImpl;
+import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStore;
+import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStoreImpl;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.hdds.utils.TransactionInfo;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.container.common.SCMTestUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Tests for {@link SCMHATransactionBufferMonitorTask} and the flush race
+ * conditions it can trigger against {@link SCMHADBTransactionBufferImpl}.
+ */
+public class TestSCMHATransactionBufferMonitorTask {
+
+ private static final long FLUSH_INTERVAL_MS = 1000L;
+ private static final TransactionInfo TRX_INFO_T4 =
+ TransactionInfo.valueOf(1, 4);
+ private static final TransactionInfo TRX_INFO_T5 =
+ TransactionInfo.valueOf(1, 5);
+
+ @TempDir
+ private File testDir;
+
+ private final AtomicLong clockMillis = new AtomicLong(0);
+ private SCMMetadataStore metadataStore;
+ private SCMHADBTransactionBufferImpl transactionBuffer;
+ private Table<String, ByteString> statefulServiceConfigTable;
+ private Table<String, TransactionInfo> transactionInfoTable;
+
+ @BeforeEach
+ public void setup() throws Exception {
+ OzoneConfiguration conf = SCMTestUtils.getConf(testDir);
+ metadataStore = new SCMMetadataStoreImpl(conf);
+ statefulServiceConfigTable = metadataStore.getStatefulServiceConfigTable();
+ transactionInfoTable = metadataStore.getTransactionInfoTable();
+
+ StorageContainerManager scm = mock(StorageContainerManager.class);
+ BlockManager blockManager = mock(BlockManager.class);
+ DeletedBlockLogImpl deletedBlockLog = mock(DeletedBlockLogImpl.class);
+ Clock clock = mock(Clock.class);
+ when(clock.millis()).thenAnswer(invocation -> clockMillis.get());
+ when(scm.getScmMetadataStore()).thenReturn(metadataStore);
+ when(scm.getSystemClock()).thenReturn(clock);
+ when(scm.getScmBlockManager()).thenReturn(blockManager);
+ when(blockManager.getDeletedBlockLog()).thenReturn(deletedBlockLog);
+
+ transactionBuffer = new SCMHADBTransactionBufferImpl(scm);
+ clockMillis.set(FLUSH_INTERVAL_MS + 1);
+ }
+
+ @AfterEach
+ public void cleanup() throws Exception {
+ if (transactionBuffer != null) {
+ transactionBuffer.close();
+ }
+ if (metadataStore != null) {
+ metadataStore.stop();
+ }
+ }
+
+ private void advanceClockPastFlushInterval() {
+ clockMillis.addAndGet(FLUSH_INTERVAL_MS + 1);
+ }
+
+ /**
+ * Demonstrates the partial flush race when shouldFlush and flush are called
+ * separately: buffered data can be committed with a stale transaction index.
+ */
+ @Test
+ public void testPartialFlushWithSeparateShouldFlushAndFlush() throws
Exception {
+ transactionBuffer.updateLatestTrxInfo(TRX_INFO_T4);
+ transactionBuffer.flush();
+
+ transactionBuffer.addToBuffer(statefulServiceConfigTable, "key",
+ ByteString.copyFromUtf8("value"));
+
+ advanceClockPastFlushInterval();
+ if (transactionBuffer.shouldFlush(FLUSH_INTERVAL_MS)) {
+ transactionBuffer.flush();
+ }
+
+ assertEquals(TRX_INFO_T4, transactionInfoTable.get(TRANSACTION_INFO_KEY));
+ assertEquals(ByteString.copyFromUtf8("value"),
+ statefulServiceConfigTable.get("key"));
+ }
+
+ @Test
+ public void testFlushIfNeededDoesNotFlushDuringTransactionApply()
+ throws Exception {
+ transactionBuffer.updateLatestTrxInfo(TRX_INFO_T4);
+ transactionBuffer.flush();
+
+ transactionBuffer.beginApplyingTransaction();
+ try {
+ transactionBuffer.addToBuffer(statefulServiceConfigTable, "key",
+ ByteString.copyFromUtf8("value"));
+ transactionBuffer.flushIfNeeded(FLUSH_INTERVAL_MS);
+ assertNull(statefulServiceConfigTable.get("key"));
+ } finally {
+ transactionBuffer.endApplyingTransaction();
+ }
+
+ transactionBuffer.updateLatestTrxInfo(TRX_INFO_T5);
+ advanceClockPastFlushInterval();
+ transactionBuffer.flushIfNeeded(FLUSH_INTERVAL_MS);
+
+ assertEquals(TRX_INFO_T5, transactionInfoTable.get(TRANSACTION_INFO_KEY));
+ assertEquals(ByteString.copyFromUtf8("value"),
+ statefulServiceConfigTable.get("key"));
+ }
+
+ /**
+ * Demonstrates that calling flush() directly inside an applyTransaction
+ * window (the old behaviour of StatefulServiceStateManagerImpl) persists
+ * the batch with the stale transaction index that was current before the
+ * apply updated it.
+ */
+ @Test
+ public void testDirectFlushDuringApplyWritesStaleTransactionInfo()
+ throws Exception {
+ transactionBuffer.updateLatestTrxInfo(TRX_INFO_T4);
+ transactionBuffer.flush();
+
+ transactionBuffer.beginApplyingTransaction();
+ try {
+ transactionBuffer.addToBuffer(statefulServiceConfigTable, "key",
+ ByteString.copyFromUtf8("value"));
+ // Old saveConfiguration behaviour: flush() before updateLatestTrxInfo.
+ transactionBuffer.flush();
+ // Data is on disk, but the transaction index is still T4 — stale.
+ assertEquals(TRX_INFO_T4,
transactionInfoTable.get(TRANSACTION_INFO_KEY));
+ assertEquals(ByteString.copyFromUtf8("value"),
+ statefulServiceConfigTable.get("key"));
+ } finally {
+ transactionBuffer.updateLatestTrxInfo(TRX_INFO_T5);
+ transactionBuffer.endApplyingTransaction();
+ }
+ }
+
+ /**
+ * Verifies that using flushIfNeeded(0) instead of flush() inside an apply
+ * window defers the write until after updateLatestTrxInfo(), keeping the
+ * on-disk transaction index consistent with the buffered data.
+ */
+ @Test
+ public void testFlushIfNeededZeroWaitDefersDuringApply() throws Exception {
+ transactionBuffer.updateLatestTrxInfo(TRX_INFO_T4);
+ transactionBuffer.flush();
+
+ transactionBuffer.beginApplyingTransaction();
+ try {
+ transactionBuffer.addToBuffer(statefulServiceConfigTable, "key",
+ ByteString.copyFromUtf8("value"));
+ // New saveConfiguration behaviour: skipped because apply is in progress.
+ transactionBuffer.flushIfNeeded(0);
+ assertNull(statefulServiceConfigTable.get("key"),
+ "flushIfNeeded must not flush while a transaction is being applied");
+ } finally {
+ transactionBuffer.updateLatestTrxInfo(TRX_INFO_T5);
+ transactionBuffer.endApplyingTransaction();
+ }
+
+ // After the apply window closes, the monitor flushes both data and the
+ // correct transaction index atomically.
+ advanceClockPastFlushInterval();
+ transactionBuffer.flushIfNeeded(FLUSH_INTERVAL_MS);
+
+ assertEquals(TRX_INFO_T5, transactionInfoTable.get(TRANSACTION_INFO_KEY));
+ assertEquals(ByteString.copyFromUtf8("value"),
+ statefulServiceConfigTable.get("key"));
+ }
+
+ @Test
+ public void testMonitorTaskDoesNotPartialFlushDuringTransactionApply()
+ throws Exception {
+ transactionBuffer.updateLatestTrxInfo(TRX_INFO_T4);
+ transactionBuffer.flush();
+
+ CountDownLatch addedToBuffer = new CountDownLatch(1);
+ CountDownLatch allowFinishApply = new CountDownLatch(1);
+ CountDownLatch applyFinished = new CountDownLatch(1);
+ SCMHATransactionBufferMonitorTask monitorTask =
+ new SCMHATransactionBufferMonitorTask(transactionBuffer,
FLUSH_INTERVAL_MS);
+
+ Thread applyThread = new Thread(() -> {
+ transactionBuffer.beginApplyingTransaction();
+ try {
+ try {
+ transactionBuffer.addToBuffer(statefulServiceConfigTable, "key",
+ ByteString.copyFromUtf8("value"));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ addedToBuffer.countDown();
+ try {
+ allowFinishApply.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ transactionBuffer.updateLatestTrxInfo(TRX_INFO_T5);
+ } finally {
+ transactionBuffer.endApplyingTransaction();
+ applyFinished.countDown();
+ }
+ });
+
+ Thread monitorThread = new Thread(() -> {
+ try {
+ while (!applyFinished.await(10, TimeUnit.MILLISECONDS)) {
+ monitorTask.run();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+
+ applyThread.start();
+ monitorThread.start();
+
+ assertTrue(addedToBuffer.await(10, TimeUnit.SECONDS),
+ "Timed out waiting for applyThread to add data to buffer");
+ monitorTask.run();
+ assertNull(statefulServiceConfigTable.get("key"),
+ "Monitor must not flush before transaction info is updated");
+
+ allowFinishApply.countDown();
+ applyThread.join(10_000);
+ monitorThread.join(10_000);
+
+ advanceClockPastFlushInterval();
+ transactionBuffer.flushIfNeeded(FLUSH_INTERVAL_MS);
+ assertEquals(TRX_INFO_T5, transactionInfoTable.get(TRANSACTION_INFO_KEY));
+ assertEquals(ByteString.copyFromUtf8("value"),
+ statefulServiceConfigTable.get("key"));
+ }
+}
diff --git
a/hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/AbstractTestStorageDistributionEndpoint.java
b/hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/AbstractTestStorageDistributionEndpoint.java
index 47f4eeb1270..70aaa01a666 100644
---
a/hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/AbstractTestStorageDistributionEndpoint.java
+++
b/hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/AbstractTestStorageDistributionEndpoint.java
@@ -21,7 +21,6 @@
import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_INTERVAL;
import static
org.apache.hadoop.hdds.HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT;
import static
org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_HA_DBTRANSACTIONBUFFER_FLUSH_INTERVAL;
-import static
org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_GAP;
import static
org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL;
import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_INTERVAL;
import static
org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_DIR_DELETING_SERVICE_INTERVAL;
@@ -146,7 +145,6 @@ protected static void initializeCluster(int numDatanodes)
throws Exception {
conf.setTimeDuration(OZONE_DIR_DELETING_SERVICE_INTERVAL, 100,
TimeUnit.MILLISECONDS);
conf.setTimeDuration(OZONE_BLOCK_DELETING_SERVICE_INTERVAL, 100,
TimeUnit.MILLISECONDS);
conf.setTimeDuration(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL, 100,
TimeUnit.MILLISECONDS);
- conf.setLong(OZONE_SCM_HA_RATIS_SNAPSHOT_GAP, 1L);
conf.setTimeDuration(HDDS_HEARTBEAT_INTERVAL, 50, TimeUnit.MILLISECONDS);
conf.setTimeDuration(HDDS_CONTAINER_REPORT_INTERVAL, 200,
TimeUnit.MILLISECONDS);
conf.setTimeDuration(OZONE_SCM_HA_DBTRANSACTIONBUFFER_FLUSH_INTERVAL, 500,
TimeUnit.MILLISECONDS);
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestDatanodeSCMNodesReconfiguration.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestDatanodeSCMNodesReconfiguration.java
index 6f38d89fb2b..eb30586ec1e 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestDatanodeSCMNodesReconfiguration.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestDatanodeSCMNodesReconfiguration.java
@@ -74,7 +74,6 @@ public void init() throws Exception {
conf.set(ScmConfigKeys.OZONE_SCM_PIPELINE_CREATION_INTERVAL, "10s");
conf.set(ScmConfigKeys.OZONE_SCM_HA_DBTRANSACTIONBUFFER_FLUSH_INTERVAL,
"5s");
- conf.set(ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_GAP, "1");
conf.setTimeDuration(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL, 100,
MILLISECONDS);
conf.setTimeDuration(HDDS_HEARTBEAT_INTERVAL, 1, SECONDS);
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestStorageContainerManagerHA.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestStorageContainerManagerHA.java
index 2825683f1ac..ae10947d52c 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestStorageContainerManagerHA.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestStorageContainerManagerHA.java
@@ -53,7 +53,6 @@ public void init() throws Exception {
conf.set(ScmConfigKeys.OZONE_SCM_PIPELINE_CREATION_INTERVAL, "10s");
conf.set(ScmConfigKeys.OZONE_SCM_HA_DBTRANSACTIONBUFFER_FLUSH_INTERVAL,
"5s");
- conf.set(ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_GAP, "1");
cluster = MiniOzoneCluster.newHABuilder(conf)
.setOMServiceId("om-service-test1")
.setSCMServiceId("scm-service-test1")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
