smengcl commented on code in PR #10830:
URL: https://github.com/apache/ozone/pull/10830#discussion_r4038757791
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntryPool.java:
##########
Review Comment:
Let's add a deterministic unit test that would repro this issue as a
regression test:
```diff
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
---
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,55 @@
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();
+ 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()));
+ BlockOutputStreamEntry closed = mock(BlockOutputStreamEntry.class);
+ when(closed.isClosed()).thenReturn(true);
+ pool.getStreamEntries().add(closed);
+
+ 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 {
+ assertTrue(allocationStarted.await(5, TimeUnit.SECONDS));
+ reader.start();
+ // Wait until the reader attempts the lookup before allowing
allocation to finish.
+ 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 {
+ finishAllocation.countDown();
+ writer.join(5000);
+ reader.join(5000);
+ }
+ assertFalse(writer.isAlive(), "Allocation should complete");
+ assertFalse(reader.isAlive(), "Lookup should complete");
+ assertSame(allocation.get(5, TimeUnit.SECONDS), read.get(5,
TimeUnit.SECONDS),
+ "Reader must not observe the intermediate index before the new
entry is added");
+ }
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
