This is an automated email from the ASF dual-hosted git repository.
smengcl 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 fe4a01b7f9c HDDS-15698. Fix concurrent write and hsync race in
BlockOutputStreamEntryPool (#10830)
fe4a01b7f9c is described below
commit fe4a01b7f9c63121330359d12d7e23f16375f791
Author: Navink <[email protected]>
AuthorDate: Fri Sep 18 07:03:03 2026 +0530
HDDS-15698. Fix concurrent write and hsync race in
BlockOutputStreamEntryPool (#10830)
---
.../client/io/BlockOutputStreamEntryPool.java | 6 +-
.../ozone/client/io/TestKeyOutputStream.java | 76 ++++++++++++++++++++++
2 files changed, 79 insertions(+), 3 deletions(-)
diff --git
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntryPool.java
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntryPool.java
index cef5a7ed056..0abfc001ee0 100644
---
a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntryPool.java
+++
b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntryPool.java
@@ -368,7 +368,7 @@ void hsyncKey(long offset) throws IOException {
}
}
- BlockOutputStreamEntry getCurrentStreamEntry() {
+ synchronized BlockOutputStreamEntry getCurrentStreamEntry() {
if (streamEntries.isEmpty() || streamEntries.size() <= currentStreamIndex)
{
return null;
} else {
@@ -407,7 +407,7 @@ long computeBufferData() {
return bufferPool.computeBufferData();
}
- void cleanup() {
+ synchronized void cleanup() {
if (excludeList != null) {
excludeList.clear();
}
@@ -432,7 +432,7 @@ public ExcludeList getExcludeList() {
return excludeList;
}
- boolean isEmpty() {
+ synchronized boolean isEmpty() {
return streamEntries.isEmpty();
}
diff --git
a/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/io/TestKeyOutputStream.java
b/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/io/TestKeyOutputStream.java
index c5b4866e4f9..798a75347fe 100644
---
a/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/io/TestKeyOutputStream.java
+++
b/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/io/TestKeyOutputStream.java
@@ -18,21 +18,40 @@
package org.apache.hadoop.ozone.client.io;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
+import org.apache.hadoop.hdds.scm.OzoneClientConfig;
+import org.apache.hadoop.hdds.scm.StreamBufferArgs;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
+import org.apache.hadoop.ozone.om.helpers.OpenKeySession;
+import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol;
import org.apache.ozone.test.GenericTestUtils;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -122,4 +141,61 @@ void testConcurrentWriteLimitOne() throws Exception {
thread2.join();
thread1.join();
}
+
+ @Test
+ void testGetCurrentStreamEntryDuringBlockAllocation() throws Exception {
+ CountDownLatch allocationStarted = new CountDownLatch(1);
+ CountDownLatch finishAllocation = new CountDownLatch(1);
+ OzoneManagerProtocol om = mock(OzoneManagerProtocol.class);
+ OmKeyLocationInfo next = new OmKeyLocationInfo.Builder()
+ .setBlockID(new BlockID(1, 2)).setLength(1024)
+ .setPipeline(mock(Pipeline.class)).build();
+ // Pause inside allocateBlock() to keep the pool in the intermediate state.
+ when(om.allocateBlock(any(), anyLong(), any())).thenAnswer(invocation -> {
+ allocationStarted.countDown();
+ assertTrue(finishAllocation.await(10, TimeUnit.SECONDS));
+ return next;
+ });
+ BlockOutputStreamEntryPool pool = new BlockOutputStreamEntryPool(new
KeyOutputStream.Builder()
+ .setConfig(new OzoneClientConfig()).setOmClient(om)
+
.setReplicationConfig(RatisReplicationConfig.getInstance(ReplicationFactor.THREE))
+ .setHandler(new OpenKeySession(1, new OmKeyInfo.Builder()
+ .setVolumeName("v").setBucketName("b").setKeyName("k").build(), 0))
+ .setStreamBufferArgs(StreamBufferArgs.Builder.getNewBuilder()
+
.setBufferSize(1024).setBufferFlushSize(1024).setBufferMaxSize(2048).build()));
+ // Add a closed entry so allocateBlockIfNeeded() must advance the index
and allocate.
+ BlockOutputStreamEntry closed = mock(BlockOutputStreamEntry.class);
+ when(closed.isClosed()).thenReturn(true);
+ pool.getStreamEntries().add(closed);
+
+ // writer allocates (holds the pool monitor) and the reader does the
concurrent lookup (as hsync).
+ FutureTask<BlockOutputStreamEntry> allocation = new FutureTask<>(() ->
pool.allocateBlockIfNeeded(false));
+ FutureTask<BlockOutputStreamEntry> read = new
FutureTask<>(pool::getCurrentStreamEntry);
+ Thread writer = new Thread(allocation, "block-allocator");
+ Thread reader = new Thread(read, "entry-reader");
+ ThreadMXBean threads = ManagementFactory.getThreadMXBean();
+ writer.start();
+ try {
+ // Wait until the writer is paused inside allocateBlock().
+ assertTrue(allocationStarted.await(5, TimeUnit.SECONDS));
+ reader.start();
+ // Deterministic sync proceed once reader is done or BLOCKED on the pool
monitor.
+ GenericTestUtils.waitFor(() -> {
+ ThreadInfo info = threads.getThreadInfo(reader.getId());
+ return read.isDone() || (info != null && info.getThreadState() ==
Thread.State.BLOCKED
+ && info.getLockOwnerId() == writer.getId()
+ && info.getLockInfo().getIdentityHashCode() ==
System.identityHashCode(pool));
+ }, 10, 5000);
+ } finally {
+ // Release allocation (appends the entry) and wait for both threads.
+ finishAllocation.countDown();
+ writer.join(5000);
+ reader.join(5000);
+ }
+ assertFalse(writer.isAlive(), "Allocation should complete");
+ assertFalse(reader.isAlive(), "Lookup should complete");
+ // Reader must see the allocated entry, not the intermediate null (fails
without the fix).
+ assertSame(allocation.get(5, TimeUnit.SECONDS), read.get(5,
TimeUnit.SECONDS),
+ "Reader must not observe the intermediate index before the new entry
is added");
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
