Copilot commented on code in PR #10633:
URL: https://github.com/apache/ozone/pull/10633#discussion_r3565426787
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java:
##########
@@ -189,23 +193,46 @@ protected List<OmKeyLocationInfo> allocateBlock(
UserInfo userInfo, OzoneManager ozoneManager)
throws IOException {
final long scmBlockSize = ozoneManager.getScmBlockSize();
+ final KeyManager keyManager = ozoneManager.getKeyManager();
int dataGroupSize = replicationConfig instanceof ECReplicationConfig
? ((ECReplicationConfig) replicationConfig).getData() : 1;
final int numBlocks = (int)
Math.min(ozoneManager.getPreallocateBlocksMax(),
(requestedSize - 1) / (scmBlockSize * dataGroupSize) + 1);
- String clientMachine = "";
- if (shouldSortDatanodes) {
- clientMachine = userInfo.getRemoteAddress();
+ final String scmClientMachine;
+ final String omClientMachine;
+ // Sorted order cached by datanode set so blocks whose pipelines share the
+ // same datanodes are sorted once (mirrors the read path's caching). Keyed
by
+ // the UUID set so it is order-insensitive and dedups across pipelines.
+ final Map<Set<String>, List<? extends DatanodeDetails>> sortedByNodes;
+ final String remoteAddress = userInfo.getRemoteAddress();
+ final NetworkTopology clusterMap = shouldSortDatanodes
+ && keyManager.isSortDatanodesForWriteEnabled()
+ ? ozoneManager.getClusterMapAllowNull() : null;
+ if (!shouldSortDatanodes) {
+ scmClientMachine = "";
+ omClientMachine = "";
+ sortedByNodes = null;
+ } else if (clusterMap != null && !remoteAddress.isEmpty()) {
+ // Sort in OM: SCM skips sorting (empty machine), OM sorts by
remoteAddress.
+ scmClientMachine = "";
+ omClientMachine = remoteAddress;
+ sortedByNodes = new HashMap<>();
+ } else {
+ // Sort in SCM (or keep order when remoteAddress is empty, since SCM
skips
+ // sorting for an empty client machine).
+ scmClientMachine = remoteAddress;
+ omClientMachine = "";
+ sortedByNodes = null;
}
List<OmKeyLocationInfo> locationInfos = new ArrayList<>(numBlocks);
String remoteUser = getRemoteUser().getShortUserName();
final List<AllocatedBlock> allocatedBlocks;
try {
allocatedBlocks =
ozoneManager.getScmClient().getBlockClient().allocateBlock(
- scmBlockSize, numBlocks, replicationConfig,
ozoneManager.getOMServiceId(), excludeList, clientMachine);
+ scmBlockSize, numBlocks, replicationConfig,
ozoneManager.getOMServiceId(), excludeList, scmClientMachine);
Review Comment:
The allocateBlock() call is on a single very long line and is likely to
exceed the project's line-length/checkstyle limit, which can break the build.
Please wrap the argument list across multiple lines.
##########
hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java:
##########
@@ -225,6 +252,241 @@ protected OMRequest doPreExecute(OMRequest
originalOMRequest)
return modifiedOmRequest;
}
+ @Test
+ public void testAllocateBlockSendsClientMachineToScmWhenFlagOff() throws
Exception {
+ // Flag off (default): OM must NOT sort; SCM receives the real client
address
+ // so it performs the sort.
+ KeyManager mockKeyManager = mock(KeyManager.class);
+ when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(false);
+ when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager);
+
+ OMAllocateBlockRequest request =
+
getOmAllocateBlockRequest(createAllocateBlockRequestWithSort("1.2.3.4"));
+ request.preExecute(ozoneManager);
+
+ ArgumentCaptor<String> clientMachine =
ArgumentCaptor.forClass(String.class);
+ verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(),
+ any(), any(), clientMachine.capture());
+ assertEquals("1.2.3.4", clientMachine.getValue());
+ verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(),
any());
+ }
+
+ @Test
+ public void testAllocateBlockDoesNotSendClientMachineToScm() throws
Exception {
+ // OM now sorts the write pipeline locally, so SCM must receive an empty
+ // clientMachine even when the client requests sorted datanodes.
+ KeyManager mockKeyManager = mock(KeyManager.class);
+ when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true);
+ when(mockKeyManager.sortDatanodesForWrite(any(), any(), any()))
+ .thenAnswer(inv -> inv.getArgument(0));
+ when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager);
+
when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class));
+
+ OMAllocateBlockRequest request =
+
getOmAllocateBlockRequest(createAllocateBlockRequestWithSort("1.2.3.4"));
+ request.preExecute(ozoneManager);
+
+ ArgumentCaptor<String> clientMachine =
ArgumentCaptor.forClass(String.class);
+ verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(),
+ any(), any(), clientMachine.capture());
+ assertEquals("", clientMachine.getValue());
+ }
+
+ @Test
+ public void testAllocateBlockFallsBackToScmWhenTopologyUnavailable() throws
Exception {
+ KeyManager mockKeyManager = mock(KeyManager.class);
+ when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true);
+ when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager);
+
+ OMAllocateBlockRequest request =
+
getOmAllocateBlockRequest(createAllocateBlockRequestWithSort("1.2.3.4"));
+ request.preExecute(ozoneManager);
+
+ ArgumentCaptor<String> clientMachine =
ArgumentCaptor.forClass(String.class);
+ verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(),
+ any(), any(), clientMachine.capture());
+ assertEquals("1.2.3.4", clientMachine.getValue());
+ verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(),
any());
+ }
+
+ @Test
+ public void testAllocateBlockSortsSharedPipelineOnce() throws Exception {
+ // Two blocks on the same 3-node pipeline must be sorted once, and the
+ // sorted order must land in every block's pipeline.
+ List<DatanodeDetails> nodes = Arrays.asList(
+ MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails());
+ Pipeline pipeline = Pipeline.newBuilder()
+ .setState(Pipeline.PipelineState.OPEN)
+ .setId(PipelineID.randomId())
+ .setReplicationConfig(
+ StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE))
+ .setNodes(nodes)
+ .build();
+ AllocatedBlock.Builder blockBuilder =
+ new AllocatedBlock.Builder().setPipeline(pipeline);
+ when(scmBlockLocationProtocol.allocateBlock(anyLong(), anyInt(), any(),
+ anyString(), any(ExcludeList.class), anyString())).thenAnswer(inv -> {
+ int num = inv.getArgument(1);
+ List<AllocatedBlock> blocks = new ArrayList<>(num);
+ for (int i = 0; i < num; i++) {
+ blockBuilder.setContainerBlockID(
+ new ContainerBlockID(CONTAINER_ID + i, LOCAL_ID + i));
+ blocks.add(blockBuilder.build());
+ }
+ return blocks;
+ });
+
+ List<DatanodeDetails> sortedOrder = new ArrayList<>(nodes);
+ Collections.reverse(sortedOrder);
+ KeyManager mockKeyManager = mock(KeyManager.class);
+ when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true);
+ when(mockKeyManager.sortDatanodesForWrite(any(), any(), any()))
+ .thenAnswer(inv -> sortedOrder);
+ when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager);
+
when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class));
+
+ OMAllocateBlockRequest request =
+ getOmAllocateBlockRequest(createAllocateBlockRequest());
+ // requestedSize spans two scmBlockSize blocks on the same pipeline.
+ List<OmKeyLocationInfo> locations =
request.allocateBlock(replicationConfig,
+ new ExcludeList(), 2 * scmBlockSize, true,
+ UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(),
ozoneManager);
+
+ // Sorted once for the shared pipeline...
+ verify(mockKeyManager, times(1)).sortDatanodesForWrite(any(),
eq("1.2.3.4"), any());
+ // ...and the sorted order is applied to every block's pipeline.
+ assertEquals(2, locations.size());
+ for (OmKeyLocationInfo location : locations) {
+ assertEquals(sortedOrder, location.getPipeline().getNodesInOrder());
+ }
+ }
+
+ @Test
+ public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws
Exception {
+ // Two pipelines share the same datanode set but in a different order. When
+ // the sort is skipped (sortDatanodesForWrite returns the input unchanged),
+ // each pipeline must keep its own order: the unsorted result must not be
+ // cached under the node set and reused for the other pipeline.
+ DatanodeDetails a = MockDatanodeDetails.randomDatanodeDetails();
+ DatanodeDetails b = MockDatanodeDetails.randomDatanodeDetails();
+ DatanodeDetails c = MockDatanodeDetails.randomDatanodeDetails();
+ List<DatanodeDetails> nodes1 = Arrays.asList(a, b, c);
+ List<DatanodeDetails> nodes2 = Arrays.asList(c, b, a);
+ Pipeline pipeline1 = Pipeline.newBuilder()
+ .setState(Pipeline.PipelineState.OPEN)
+ .setId(PipelineID.randomId())
+ .setReplicationConfig(
+ StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE))
+ .setNodes(nodes1)
+ .build();
+ Pipeline pipeline2 = Pipeline.newBuilder()
+ .setState(Pipeline.PipelineState.OPEN)
+ .setId(PipelineID.randomId())
+ .setReplicationConfig(
+ StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE))
+ .setNodes(nodes2)
+ .build();
+ AllocatedBlock block1 = new AllocatedBlock.Builder().setPipeline(pipeline1)
+ .setContainerBlockID(new ContainerBlockID(CONTAINER_ID,
LOCAL_ID)).build();
+ AllocatedBlock block2 = new AllocatedBlock.Builder().setPipeline(pipeline2)
+ .setContainerBlockID(new ContainerBlockID(CONTAINER_ID + 1, LOCAL_ID +
1)).build();
+ when(scmBlockLocationProtocol.allocateBlock(anyLong(), anyInt(), any(),
+ anyString(), any(ExcludeList.class), anyString()))
+ .thenReturn(Arrays.asList(block1, block2));
+
+ KeyManager mockKeyManager = mock(KeyManager.class);
+ when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true);
+ // Skip the sort: return the input list instance unchanged.
+ when(mockKeyManager.sortDatanodesForWrite(any(), any(), any()))
+ .thenAnswer(inv -> inv.getArgument(0));
+ when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager);
+
when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class));
+
+ OMAllocateBlockRequest request =
+ getOmAllocateBlockRequest(createAllocateBlockRequest());
+ List<OmKeyLocationInfo> locations =
request.allocateBlock(replicationConfig,
+ new ExcludeList(), 2 * scmBlockSize, true,
+ UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(),
ozoneManager);
+
+ assertEquals(2, locations.size());
+ // Each pipeline keeps its own order; the skipped-sort result is not
shared.
+ assertEquals(nodes1, locations.get(0).getPipeline().getNodesInOrder());
+ assertEquals(nodes2, locations.get(1).getPipeline().getNodesInOrder());
+ // Sorted per pipeline, since the unsorted result is not cached.
+ verify(mockKeyManager, times(2)).sortDatanodesForWrite(any(),
eq("1.2.3.4"), any());
+ }
+
+ @Test
+ public void testAllocateBlockKeepsOrderWhenRemoteAddressEmpty() throws
Exception {
+ // Sort enabled and topology available, but the client has no remote
address:
+ // OM must not sort, SCM receives an empty clientMachine, and the pipeline
+ // order is preserved.
+ List<DatanodeDetails> nodes = Arrays.asList(
+ MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails());
+ Pipeline pipeline = Pipeline.newBuilder()
+ .setState(Pipeline.PipelineState.OPEN)
+ .setId(PipelineID.randomId())
+ .setReplicationConfig(
+ StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE))
+ .setNodes(nodes)
+ .build();
+ AllocatedBlock block = new AllocatedBlock.Builder().setPipeline(pipeline)
+ .setContainerBlockID(new ContainerBlockID(CONTAINER_ID,
LOCAL_ID)).build();
+ ArgumentCaptor<String> clientMachine =
ArgumentCaptor.forClass(String.class);
+ when(scmBlockLocationProtocol.allocateBlock(anyLong(), anyInt(), any(),
+ anyString(), any(ExcludeList.class), clientMachine.capture()))
+ .thenReturn(Collections.singletonList(block));
+
+ KeyManager mockKeyManager = mock(KeyManager.class);
+ when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true);
+ when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager);
+
when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class));
+
+ OMAllocateBlockRequest request =
+ getOmAllocateBlockRequest(createAllocateBlockRequest());
+ List<OmKeyLocationInfo> locations =
request.allocateBlock(replicationConfig,
+ new ExcludeList(), scmBlockSize, true,
+ UserInfo.newBuilder().setRemoteAddress("").build(), ozoneManager);
+
+ assertEquals("", clientMachine.getValue());
+ verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(),
any());
+ assertEquals(1, locations.size());
+ // Assert the write order (nodesInOrder), which copyWithNodesInOrder would
+ // have changed had OM sorted; it must stay as the original pipeline order.
+ assertEquals(nodes, locations.get(0).getPipeline().getNodesInOrder());
+ }
+
+ @Test
+ public void sortDatanodesForWriteRequiresClientMachine() {
+ List<DatanodeDetails> nodes = Arrays.asList(
+ MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails(),
+ MockDatanodeDetails.randomDatanodeDetails());
+ assertThrows(IllegalArgumentException.class,
+ () -> keyManager.sortDatanodesForWrite(nodes, "",
mock(NetworkTopology.class)));
+ }
+
+ // Like createAllocateBlockRequest, but sets sortDatanodes and a UserInfo
remote address for OM-side sort.
Review Comment:
This comment line exceeds typical checkstyle line-length limits and may fail
formatting checks. Please wrap it onto multiple lines.
--
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]
