From 06c1508fa7ece2f12b1fffb98a31b03774af68f5 Mon Sep 17 00:00:00 2001 From: Chia-Chuan Yu Date: Mon, 17 Mar 2025 19:58:46 +0800 Subject: [PATCH 01/10] HDDS-12604. Reduce duplication in TestContainerStateMachine --- .../ratis/TestContainerStateMachine.java | 91 +++++++++---------- 1 file changed, 45 insertions(+), 46 deletions(-) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java index 22a335e1594e..393212e0d4ab 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java @@ -126,6 +126,37 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio TransactionContext trx = mock(TransactionContext.class); ContainerStateMachine.Context context = mock(ContainerStateMachine.Context.class); when(trx.getStateMachineContext()).thenReturn(context); + + setUpMockDispatcherReturn(failWithException); + setUpMockRequestProto(context); + + AtomicReference throwable = new AtomicReference<>(null); + Function throwableSetter = getThrowableSetter(throwable); + + stateMachine.write(entry, trx).exceptionally(throwableSetter).get(); + verify(dispatcher, times(1)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), + any(DispatcherContext.class)); + reset(dispatcher); + assertNotNull(throwable.get()); + assertResults(failWithException, throwable); + + // Writing data to another container(containerId 2) should also fail. + when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() + .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( + ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8("Test Data")) + .setBlockID( + ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(2).setLocalID(1).build()).build()) + .setContainerID(2) + .setDatanodeUuid(UUID.randomUUID().toString()).build()); + stateMachine.write(entry, trx).exceptionally(throwableSetter).get(); + verify(dispatcher, times(0)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), + any(DispatcherContext.class)); + assertInstanceOf(StorageContainerException.class, throwable.get()); + StorageContainerException sce = (StorageContainerException) throwable.get(); + assertEquals(ContainerProtos.Result.CONTAINER_UNHEALTHY, sce.getResult()); + } + + public final void setUpMockDispatcherReturn(boolean failWithException) { if (failWithException) { when(dispatcher.dispatch(any(), any())).thenThrow(new RuntimeException()); } else { @@ -134,7 +165,9 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio .setResult(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR) .build()); } + } + public final void setUpMockRequestProto(ContainerStateMachine.Context context) { when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8("Test Data")) @@ -142,16 +175,17 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(1).setLocalID(1).build()).build()) .setContainerID(1) .setDatanodeUuid(UUID.randomUUID().toString()).build()); - AtomicReference throwable = new AtomicReference<>(null); + } + + public final Function getThrowableSetter(AtomicReference throwable) { Function throwableSetter = t -> { throwable.set(t); return null; }; - stateMachine.write(entry, trx).exceptionally(throwableSetter).get(); - verify(dispatcher, times(1)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), - any(DispatcherContext.class)); - reset(dispatcher); - assertNotNull(throwable.get()); + return throwableSetter; + } + + public final void assertResults(boolean failWithException, AtomicReference throwable) { if (failWithException) { assertInstanceOf(RuntimeException.class, throwable.get()); } else { @@ -159,20 +193,6 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio StorageContainerException sce = (StorageContainerException) throwable.get(); assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, sce.getResult()); } - // Writing data to another container(containerId 2) should also fail. - when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() - .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8("Test Data")) - .setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(2).setLocalID(1).build()).build()) - .setContainerID(2) - .setDatanodeUuid(UUID.randomUUID().toString()).build()); - stateMachine.write(entry, trx).exceptionally(throwableSetter).get(); - verify(dispatcher, times(0)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), - any(DispatcherContext.class)); - assertInstanceOf(StorageContainerException.class, throwable.get()); - StorageContainerException sce = (StorageContainerException) throwable.get(); - assertEquals(ContainerProtos.Result.CONTAINER_UNHEALTHY, sce.getResult()); } @ParameterizedTest @@ -186,14 +206,8 @@ public void testApplyTransactionFailure(boolean failWithException) throws Execut ContainerStateMachine.Context context = mock(ContainerStateMachine.Context.class); when(trx.getLogEntry()).thenReturn(entry); when(trx.getStateMachineContext()).thenReturn(context); - if (failWithException) { - when(dispatcher.dispatch(any(), any())).thenThrow(new RuntimeException()); - } else { - when(dispatcher.dispatch(any(), any())).thenReturn(ContainerProtos.ContainerCommandResponseProto - .newBuilder().setCmdType(ContainerProtos.Type.WriteChunk) - .setResult(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR) - .build()); - } + + setUpMockDispatcherReturn(failWithException); // Failing apply transaction on congtainer 1. when(context.getLogProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( @@ -202,10 +216,7 @@ public void testApplyTransactionFailure(boolean failWithException) throws Execut .setContainerID(1) .setDatanodeUuid(UUID.randomUUID().toString()).build()); AtomicReference throwable = new AtomicReference<>(null); - Function throwableSetter = t -> { - throwable.set(t); - return null; - }; + Function throwableSetter = getThrowableSetter(throwable); //apply transaction will fail because of runtime exception thrown by dispatcher, which marks the first // failure on container 1. stateMachine.applyTransaction(trx).exceptionally(throwableSetter).get(); @@ -213,13 +224,7 @@ public void testApplyTransactionFailure(boolean failWithException) throws Execut any(DispatcherContext.class)); reset(dispatcher); assertNotNull(throwable.get()); - if (failWithException) { - assertInstanceOf(RuntimeException.class, throwable.get()); - } else { - assertInstanceOf(StorageContainerException.class, throwable.get()); - StorageContainerException sce = (StorageContainerException) throwable.get(); - assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, sce.getResult()); - } + assertResults(failWithException, throwable); // Another apply transaction on same container 1 should fail because the previous apply transaction failed. stateMachine.applyTransaction(trx).exceptionally(throwableSetter).get(); verify(dispatcher, times(0)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), @@ -272,13 +277,7 @@ public void testWriteTimout() throws Exception { return null; }).when(dispatcher).dispatch(any(), any()); - when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() - .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8("Test Data")) - .setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(1).setLocalID(1).build()).build()) - .setContainerID(1) - .setDatanodeUuid(UUID.randomUUID().toString()).build()); + setUpMockRequestProto(context); AtomicReference throwable = new AtomicReference<>(null); Function throwableSetter = t -> { throwable.set(t); From cca89ed9ce289739dea54140743a6e7b2f179182 Mon Sep 17 00:00:00 2001 From: Chia-Chuan Yu Date: Mon, 17 Mar 2025 20:02:21 +0800 Subject: [PATCH 02/10] substitute with function --- .../transport/server/ratis/TestContainerStateMachine.java | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java index 393212e0d4ab..598e3f75f03f 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java @@ -279,10 +279,7 @@ public void testWriteTimout() throws Exception { setUpMockRequestProto(context); AtomicReference throwable = new AtomicReference<>(null); - Function throwableSetter = t -> { - throwable.set(t); - return null; - }; + Function throwableSetter = getThrowableSetter(throwable); Field writeChunkWaitMaxNs = stateMachine.getClass().getDeclaredField("writeChunkWaitMaxNs"); writeChunkWaitMaxNs.setAccessible(true); writeChunkWaitMaxNs.set(stateMachine, 1000_000_000); From f74bd2bc50264448165f39328e378141736022e7 Mon Sep 17 00:00:00 2001 From: Chia-Chuan Yu Date: Mon, 17 Mar 2025 21:34:39 +0800 Subject: [PATCH 03/10] Refactor functions --- .../ratis/TestContainerStateMachine.java | 49 +++++++++---------- 1 file changed, 22 insertions(+), 27 deletions(-) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java index 598e3f75f03f..5dafd363c0da 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java @@ -128,7 +128,7 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio when(trx.getStateMachineContext()).thenReturn(context); setUpMockDispatcherReturn(failWithException); - setUpMockRequestProto(context); + setUpMockRequestProtoReturn(context, "Test Data", 1, 1); AtomicReference throwable = new AtomicReference<>(null); Function throwableSetter = getThrowableSetter(throwable); @@ -141,13 +141,8 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio assertResults(failWithException, throwable); // Writing data to another container(containerId 2) should also fail. - when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() - .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8("Test Data")) - .setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(2).setLocalID(1).build()).build()) - .setContainerID(2) - .setDatanodeUuid(UUID.randomUUID().toString()).build()); + setUpMockRequestProtoReturn(context, "Test Data", 2, 1); + stateMachine.write(entry, trx).exceptionally(throwableSetter).get(); verify(dispatcher, times(0)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), any(DispatcherContext.class)); @@ -167,22 +162,31 @@ public final void setUpMockDispatcherReturn(boolean failWithException) { } } - public final void setUpMockRequestProto(ContainerStateMachine.Context context) { + public final void setUpMockRequestProtoReturn(ContainerStateMachine.Context context, String content, + int containerId, int localId) { when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8("Test Data")) + ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8(content)) .setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(1).setLocalID(1).build()).build()) - .setContainerID(1) + ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(containerId).setLocalID(localId).build()).build()) + .setContainerID(containerId) + .setDatanodeUuid(UUID.randomUUID().toString()).build()); + } + + public final void setUpMockLogProtoReturn(ContainerStateMachine.Context context, int containerId, int localId) { + when(context.getLogProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() + .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( + ContainerProtos.WriteChunkRequestProto.newBuilder().setBlockID( + ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(containerId).setLocalID(localId).build()).build()) + .setContainerID(containerId) .setDatanodeUuid(UUID.randomUUID().toString()).build()); } public final Function getThrowableSetter(AtomicReference throwable) { - Function throwableSetter = t -> { + return t -> { throwable.set(t); return null; }; - return throwableSetter; } public final void assertResults(boolean failWithException, AtomicReference throwable) { @@ -209,12 +213,8 @@ public void testApplyTransactionFailure(boolean failWithException) throws Execut setUpMockDispatcherReturn(failWithException); // Failing apply transaction on congtainer 1. - when(context.getLogProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() - .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(1).setLocalID(1).build()).build()) - .setContainerID(1) - .setDatanodeUuid(UUID.randomUUID().toString()).build()); + setUpMockLogProtoReturn(context, 1, 1); + AtomicReference throwable = new AtomicReference<>(null); Function throwableSetter = getThrowableSetter(throwable); //apply transaction will fail because of runtime exception thrown by dispatcher, which marks the first @@ -235,12 +235,7 @@ public void testApplyTransactionFailure(boolean failWithException) throws Execut // Another apply transaction on a different container 2 shouldn't fail because the previous apply transaction // failure was only on container 1. - when(context.getLogProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() - .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(2).setLocalID(1).build()).build()) - .setContainerID(2) - .setDatanodeUuid(UUID.randomUUID().toString()).build()); + setUpMockLogProtoReturn(context, 2, 1); reset(dispatcher); throwable.set(null); @@ -277,7 +272,7 @@ public void testWriteTimout() throws Exception { return null; }).when(dispatcher).dispatch(any(), any()); - setUpMockRequestProto(context); + setUpMockRequestProtoReturn(context, "Test Data", 1, 1); AtomicReference throwable = new AtomicReference<>(null); Function throwableSetter = getThrowableSetter(throwable); Field writeChunkWaitMaxNs = stateMachine.getClass().getDeclaredField("writeChunkWaitMaxNs"); From 4bc69ef4320253140fc3458e20cd47e690c2f9bf Mon Sep 17 00:00:00 2001 From: Chia-Chuan Yu Date: Mon, 17 Mar 2025 21:38:06 +0800 Subject: [PATCH 04/10] Fixed styles --- .../transport/server/ratis/TestContainerStateMachine.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java index 5dafd363c0da..9e3d9cdd20df 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java @@ -168,7 +168,8 @@ public final void setUpMockRequestProtoReturn(ContainerStateMachine.Context cont .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8(content)) .setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(containerId).setLocalID(localId).build()).build()) + ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(containerId) + .setLocalID(localId).build()).build()) .setContainerID(containerId) .setDatanodeUuid(UUID.randomUUID().toString()).build()); } @@ -177,7 +178,8 @@ public final void setUpMockLogProtoReturn(ContainerStateMachine.Context context, when(context.getLogProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( ContainerProtos.WriteChunkRequestProto.newBuilder().setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(containerId).setLocalID(localId).build()).build()) + ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(containerId) + .setLocalID(localId).build()).build()) .setContainerID(containerId) .setDatanodeUuid(UUID.randomUUID().toString()).build()); } From 07394a5f0243cbb76c68896a59cce39d82344ebe Mon Sep 17 00:00:00 2001 From: Chia-Chuan Yu Date: Mon, 17 Mar 2025 19:58:46 +0800 Subject: [PATCH 05/10] HDDS-12604. Reduce duplication in TestContainerStateMachine --- .../ratis/TestContainerStateMachine.java | 91 +++++++++---------- 1 file changed, 45 insertions(+), 46 deletions(-) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java index a0c543d72325..20d45287767f 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java @@ -129,6 +129,37 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio TransactionContext trx = mock(TransactionContext.class); ContainerStateMachine.Context context = mock(ContainerStateMachine.Context.class); when(trx.getStateMachineContext()).thenReturn(context); + + setUpMockDispatcherReturn(failWithException); + setUpMockRequestProto(context); + + AtomicReference throwable = new AtomicReference<>(null); + Function throwableSetter = getThrowableSetter(throwable); + + stateMachine.write(entry, trx).exceptionally(throwableSetter).get(); + verify(dispatcher, times(1)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), + any(DispatcherContext.class)); + reset(dispatcher); + assertNotNull(throwable.get()); + assertResults(failWithException, throwable); + + // Writing data to another container(containerId 2) should also fail. + when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() + .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( + ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8("Test Data")) + .setBlockID( + ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(2).setLocalID(1).build()).build()) + .setContainerID(2) + .setDatanodeUuid(UUID.randomUUID().toString()).build()); + stateMachine.write(entryNext, trx).exceptionally(throwableSetter).get(); + verify(dispatcher, times(0)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), + any(DispatcherContext.class)); + assertInstanceOf(StorageContainerException.class, throwable.get()); + StorageContainerException sce = (StorageContainerException) throwable.get(); + assertEquals(ContainerProtos.Result.CONTAINER_UNHEALTHY, sce.getResult()); + } + + public final void setUpMockDispatcherReturn(boolean failWithException) { if (failWithException) { when(dispatcher.dispatch(any(), any())).thenThrow(new RuntimeException()); } else { @@ -137,7 +168,9 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio .setResult(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR) .build()); } + } + public final void setUpMockRequestProto(ContainerStateMachine.Context context) { when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8("Test Data")) @@ -145,16 +178,17 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(1).setLocalID(1).build()).build()) .setContainerID(1) .setDatanodeUuid(UUID.randomUUID().toString()).build()); - AtomicReference throwable = new AtomicReference<>(null); + } + + public final Function getThrowableSetter(AtomicReference throwable) { Function throwableSetter = t -> { throwable.set(t); return null; }; - stateMachine.write(entry, trx).exceptionally(throwableSetter).get(); - verify(dispatcher, times(1)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), - any(DispatcherContext.class)); - reset(dispatcher); - assertNotNull(throwable.get()); + return throwableSetter; + } + + public final void assertResults(boolean failWithException, AtomicReference throwable) { if (failWithException) { assertInstanceOf(RuntimeException.class, throwable.get()); } else { @@ -162,20 +196,6 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio StorageContainerException sce = (StorageContainerException) throwable.get(); assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, sce.getResult()); } - // Writing data to another container(containerId 2) should also fail. - when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() - .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8("Test Data")) - .setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(2).setLocalID(1).build()).build()) - .setContainerID(2) - .setDatanodeUuid(UUID.randomUUID().toString()).build()); - stateMachine.write(entryNext, trx).exceptionally(throwableSetter).get(); - verify(dispatcher, times(0)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), - any(DispatcherContext.class)); - assertInstanceOf(StorageContainerException.class, throwable.get()); - StorageContainerException sce = (StorageContainerException) throwable.get(); - assertEquals(ContainerProtos.Result.CONTAINER_UNHEALTHY, sce.getResult()); } @ParameterizedTest @@ -189,14 +209,8 @@ public void testApplyTransactionFailure(boolean failWithException) throws Execut ContainerStateMachine.Context context = mock(ContainerStateMachine.Context.class); when(trx.getLogEntry()).thenReturn(entry); when(trx.getStateMachineContext()).thenReturn(context); - if (failWithException) { - when(dispatcher.dispatch(any(), any())).thenThrow(new RuntimeException()); - } else { - when(dispatcher.dispatch(any(), any())).thenReturn(ContainerProtos.ContainerCommandResponseProto - .newBuilder().setCmdType(ContainerProtos.Type.WriteChunk) - .setResult(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR) - .build()); - } + + setUpMockDispatcherReturn(failWithException); // Failing apply transaction on congtainer 1. when(context.getLogProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( @@ -205,10 +219,7 @@ public void testApplyTransactionFailure(boolean failWithException) throws Execut .setContainerID(1) .setDatanodeUuid(UUID.randomUUID().toString()).build()); AtomicReference throwable = new AtomicReference<>(null); - Function throwableSetter = t -> { - throwable.set(t); - return null; - }; + Function throwableSetter = getThrowableSetter(throwable); //apply transaction will fail because of runtime exception thrown by dispatcher, which marks the first // failure on container 1. stateMachine.applyTransaction(trx).exceptionally(throwableSetter).get(); @@ -216,13 +227,7 @@ public void testApplyTransactionFailure(boolean failWithException) throws Execut any(DispatcherContext.class)); reset(dispatcher); assertNotNull(throwable.get()); - if (failWithException) { - assertInstanceOf(RuntimeException.class, throwable.get()); - } else { - assertInstanceOf(StorageContainerException.class, throwable.get()); - StorageContainerException sce = (StorageContainerException) throwable.get(); - assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, sce.getResult()); - } + assertResults(failWithException, throwable); // Another apply transaction on same container 1 should fail because the previous apply transaction failed. stateMachine.applyTransaction(trx).exceptionally(throwableSetter).get(); verify(dispatcher, times(0)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), @@ -275,13 +280,7 @@ public void testWriteTimout() throws Exception { return null; }).when(dispatcher).dispatch(any(), any()); - when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() - .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8("Test Data")) - .setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(1).setLocalID(1).build()).build()) - .setContainerID(1) - .setDatanodeUuid(UUID.randomUUID().toString()).build()); + setUpMockRequestProto(context); AtomicReference throwable = new AtomicReference<>(null); Function throwableSetter = t -> { throwable.set(t); From 44ff1b3edf5864dd89b5f6fc64fd3de0974fbdd9 Mon Sep 17 00:00:00 2001 From: Chia-Chuan Yu Date: Tue, 18 Mar 2025 20:26:49 +0800 Subject: [PATCH 06/10] refactor functions --- .../ratis/TestContainerStateMachine.java | 47 +++++++++---------- 1 file changed, 21 insertions(+), 26 deletions(-) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java index 20d45287767f..c20234565e61 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java @@ -131,7 +131,7 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio when(trx.getStateMachineContext()).thenReturn(context); setUpMockDispatcherReturn(failWithException); - setUpMockRequestProto(context); + setUpMockRequestProtoReturn(context, "Test Data", 1, 1); AtomicReference throwable = new AtomicReference<>(null); Function throwableSetter = getThrowableSetter(throwable); @@ -144,13 +144,7 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio assertResults(failWithException, throwable); // Writing data to another container(containerId 2) should also fail. - when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() - .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8("Test Data")) - .setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(2).setLocalID(1).build()).build()) - .setContainerID(2) - .setDatanodeUuid(UUID.randomUUID().toString()).build()); + setUpMockRequestProtoReturn(context, "Test Data", 2, 1); stateMachine.write(entryNext, trx).exceptionally(throwableSetter).get(); verify(dispatcher, times(0)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), any(DispatcherContext.class)); @@ -170,13 +164,15 @@ public final void setUpMockDispatcherReturn(boolean failWithException) { } } - public final void setUpMockRequestProto(ContainerStateMachine.Context context) { + public final void setUpMockRequestProtoReturn(ContainerStateMachine.Context context, String content, + int containerId, int localId) { when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8("Test Data")) + ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8(content)) .setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(1).setLocalID(1).build()).build()) - .setContainerID(1) + ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(containerId) + .setLocalID(localId).build()).build()) + .setContainerID(containerId) .setDatanodeUuid(UUID.randomUUID().toString()).build()); } @@ -198,6 +194,16 @@ public final void assertResults(boolean failWithException, AtomicReference throwable = new AtomicReference<>(null); Function throwableSetter = getThrowableSetter(throwable); //apply transaction will fail because of runtime exception thrown by dispatcher, which marks the first @@ -238,13 +239,7 @@ public void testApplyTransactionFailure(boolean failWithException) throws Execut // Another apply transaction on a different container 2 shouldn't fail because the previous apply transaction // failure was only on container 1. - when(context.getLogProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() - .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(2).setLocalID(1).build()).build()) - .setContainerID(2) - .setDatanodeUuid(UUID.randomUUID().toString()).build()); - + setUpLogProtoReturn(context, 2, 1); reset(dispatcher); throwable.set(null); when(dispatcher.dispatch(any(), any())).thenReturn(ContainerProtos.ContainerCommandResponseProto @@ -280,7 +275,7 @@ public void testWriteTimout() throws Exception { return null; }).when(dispatcher).dispatch(any(), any()); - setUpMockRequestProto(context); + setUpMockRequestProtoReturn(context, "Test data", 1, 1); AtomicReference throwable = new AtomicReference<>(null); Function throwableSetter = t -> { throwable.set(t); From 53e8207c9e3a04ff1c44516d0a685a6698b642a3 Mon Sep 17 00:00:00 2001 From: Chia-Chuan Yu Date: Thu, 20 Mar 2025 17:47:41 +0800 Subject: [PATCH 07/10] Apply comments --- .../ratis/TestContainerStateMachine.java | 187 +++++++++--------- 1 file changed, 98 insertions(+), 89 deletions(-) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java index c20234565e61..cf264b26f85b 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java @@ -51,7 +51,6 @@ import org.apache.ratis.proto.RaftProtos; import org.apache.ratis.protocol.Message; import org.apache.ratis.protocol.RaftGroup; -import org.apache.ratis.protocol.RaftGroupId; import org.apache.ratis.protocol.RaftPeer; import org.apache.ratis.server.DivisionInfo; import org.apache.ratis.server.RaftServer; @@ -79,6 +78,7 @@ abstract class TestContainerStateMachine { .setNameFormat("ChunkWriter-" + i + "-%d") .build())).collect(Collectors.toList()); private final boolean isLeader; + private final String CONTAINER_DATA = "Test Data"; TestContainerStateMachine(boolean isLeader) { this.isLeader = isLeader; @@ -101,8 +101,6 @@ public void setup() throws IOException { when(division.getInfo()).thenReturn(info); when(info.isLeader()).thenReturn(isLeader); when(ratisServer.getServerDivision(any())).thenReturn(division); - stateMachine = new ContainerStateMachine(null, - RaftGroupId.randomId(), dispatcher, controller, executor, ratisServer, conf, "containerOp"); } @@ -131,79 +129,27 @@ public void testWriteFailure(boolean failWithException) throws ExecutionExceptio when(trx.getStateMachineContext()).thenReturn(context); setUpMockDispatcherReturn(failWithException); - setUpMockRequestProtoReturn(context, "Test Data", 1, 1); + setUpMockRequestProtoReturn(context, 1, 1); - AtomicReference throwable = new AtomicReference<>(null); - Function throwableSetter = getThrowableSetter(throwable); + ThrowableCatcher catcher = new ThrowableCatcher(); - stateMachine.write(entry, trx).exceptionally(throwableSetter).get(); + stateMachine.write(entry, trx).exceptionally(catcher.asSetter()).get(); verify(dispatcher, times(1)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), any(DispatcherContext.class)); reset(dispatcher); - assertNotNull(throwable.get()); - assertResults(failWithException, throwable); + assertNotNull(catcher.getReceived()); + assertResults(failWithException, catcher.getCaught()); // Writing data to another container(containerId 2) should also fail. - setUpMockRequestProtoReturn(context, "Test Data", 2, 1); - stateMachine.write(entryNext, trx).exceptionally(throwableSetter).get(); + setUpMockRequestProtoReturn(context, 2, 1); + stateMachine.write(entryNext, trx).exceptionally(catcher.asSetter()).get(); verify(dispatcher, times(0)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), any(DispatcherContext.class)); - assertInstanceOf(StorageContainerException.class, throwable.get()); - StorageContainerException sce = (StorageContainerException) throwable.get(); + assertInstanceOf(StorageContainerException.class, catcher.getReceived()); + StorageContainerException sce = (StorageContainerException) catcher.getReceived(); assertEquals(ContainerProtos.Result.CONTAINER_UNHEALTHY, sce.getResult()); } - public final void setUpMockDispatcherReturn(boolean failWithException) { - if (failWithException) { - when(dispatcher.dispatch(any(), any())).thenThrow(new RuntimeException()); - } else { - when(dispatcher.dispatch(any(), any())).thenReturn(ContainerProtos.ContainerCommandResponseProto - .newBuilder().setCmdType(ContainerProtos.Type.WriteChunk) - .setResult(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR) - .build()); - } - } - - public final void setUpMockRequestProtoReturn(ContainerStateMachine.Context context, String content, - int containerId, int localId) { - when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() - .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8(content)) - .setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(containerId) - .setLocalID(localId).build()).build()) - .setContainerID(containerId) - .setDatanodeUuid(UUID.randomUUID().toString()).build()); - } - - public final Function getThrowableSetter(AtomicReference throwable) { - Function throwableSetter = t -> { - throwable.set(t); - return null; - }; - return throwableSetter; - } - - public final void assertResults(boolean failWithException, AtomicReference throwable) { - if (failWithException) { - assertInstanceOf(RuntimeException.class, throwable.get()); - } else { - assertInstanceOf(StorageContainerException.class, throwable.get()); - StorageContainerException sce = (StorageContainerException) throwable.get(); - assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, sce.getResult()); - } - } - - public final void setUpLogProtoReturn(ContainerStateMachine.Context context, int containerId, int localId) { - when(context.getLogProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() - .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( - ContainerProtos.WriteChunkRequestProto.newBuilder().setBlockID( - ContainerProtos.DatanodeBlockID.newBuilder(). - setContainerID(containerId).setLocalID(localId).build()).build()) - .setContainerID(containerId) - .setDatanodeUuid(UUID.randomUUID().toString()).build()); - } - @ParameterizedTest @ValueSource(booleans = {true, false}) public void testApplyTransactionFailure(boolean failWithException) throws ExecutionException, @@ -219,36 +165,35 @@ public void testApplyTransactionFailure(boolean failWithException) throws Execut setUpMockDispatcherReturn(failWithException); // Failing apply transaction on congtainer 1. setUpLogProtoReturn(context, 1, 1); - AtomicReference throwable = new AtomicReference<>(null); - Function throwableSetter = getThrowableSetter(throwable); + ThrowableCatcher catcher = new ThrowableCatcher(); //apply transaction will fail because of runtime exception thrown by dispatcher, which marks the first // failure on container 1. - stateMachine.applyTransaction(trx).exceptionally(throwableSetter).get(); + stateMachine.applyTransaction(trx).exceptionally(catcher.asSetter()).get(); verify(dispatcher, times(1)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), any(DispatcherContext.class)); reset(dispatcher); - assertNotNull(throwable.get()); - assertResults(failWithException, throwable); + assertNotNull(catcher.getCaught()); + assertResults(failWithException, catcher.getCaught()); // Another apply transaction on same container 1 should fail because the previous apply transaction failed. - stateMachine.applyTransaction(trx).exceptionally(throwableSetter).get(); + stateMachine.applyTransaction(trx).exceptionally(catcher.asSetter()).get(); verify(dispatcher, times(0)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), any(DispatcherContext.class)); - assertInstanceOf(StorageContainerException.class, throwable.get()); - StorageContainerException sce = (StorageContainerException) throwable.get(); + assertInstanceOf(StorageContainerException.class, catcher.getReceived()); + StorageContainerException sce = (StorageContainerException) catcher.getReceived(); assertEquals(ContainerProtos.Result.CONTAINER_UNHEALTHY, sce.getResult()); // Another apply transaction on a different container 2 shouldn't fail because the previous apply transaction // failure was only on container 1. setUpLogProtoReturn(context, 2, 1); reset(dispatcher); - throwable.set(null); + catcher.getCaught().set(null); when(dispatcher.dispatch(any(), any())).thenReturn(ContainerProtos.ContainerCommandResponseProto .newBuilder().setCmdType(ContainerProtos.Type.WriteChunk).setResult(ContainerProtos.Result.SUCCESS) .build()); - Message succcesfulTransaction = stateMachine.applyTransaction(trx).exceptionally(throwableSetter).get(); + Message succcesfulTransaction = stateMachine.applyTransaction(trx).exceptionally(catcher.asSetter()).get(); verify(dispatcher, times(1)).dispatch(any(ContainerProtos.ContainerCommandRequestProto.class), any(DispatcherContext.class)); - assertNull(throwable.get()); + assertNull(catcher.getReceived()); ContainerProtos.ContainerCommandResponseProto resp = ContainerProtos.ContainerCommandResponseProto.parseFrom(succcesfulTransaction.getContent()); assertEquals(ContainerProtos.Result.SUCCESS, resp.getResult()); @@ -275,26 +220,90 @@ public void testWriteTimout() throws Exception { return null; }).when(dispatcher).dispatch(any(), any()); - setUpMockRequestProtoReturn(context, "Test data", 1, 1); - AtomicReference throwable = new AtomicReference<>(null); - Function throwableSetter = t -> { - throwable.set(t); - return null; - }; + setUpMockRequestProtoReturn(context, 1, 1); + ThrowableCatcher catcher = new ThrowableCatcher(); + Field writeChunkWaitMaxNs = stateMachine.getClass().getDeclaredField("writeChunkWaitMaxNs"); writeChunkWaitMaxNs.setAccessible(true); writeChunkWaitMaxNs.set(stateMachine, 1000_000_000); CompletableFuture firstWrite = stateMachine.write(entry, trx); Thread.sleep(2000); CompletableFuture secondWrite = stateMachine.write(entryNext, trx); - firstWrite.exceptionally(throwableSetter).get(); - assertNotNull(throwable.get()); - assertInstanceOf(InterruptedException.class, throwable.get()); - - secondWrite.exceptionally(throwableSetter).get(); - assertNotNull(throwable.get()); - assertInstanceOf(StorageContainerException.class, throwable.get()); - StorageContainerException sce = (StorageContainerException) throwable.get(); + firstWrite.exceptionally(catcher.asSetter()).get(); + assertNotNull(catcher.getCaught()); + assertInstanceOf(InterruptedException.class, catcher.getReceived()); + + secondWrite.exceptionally(catcher.asSetter()).get(); + assertNotNull(catcher.getReceived()); + assertInstanceOf(StorageContainerException.class, catcher.getReceived()); + StorageContainerException sce = (StorageContainerException) catcher.getReceived(); assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, sce.getResult()); } + + private void setUpMockDispatcherReturn(boolean failWithException) { + if (failWithException) { + when(dispatcher.dispatch(any(), any())).thenThrow(new RuntimeException()); + } else { + when(dispatcher.dispatch(any(), any())).thenReturn(ContainerProtos.ContainerCommandResponseProto + .newBuilder().setCmdType(ContainerProtos.Type.WriteChunk) + .setResult(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR) + .build()); + } + } + + private void setUpMockRequestProtoReturn(ContainerStateMachine.Context context, + int containerId, int localId) { + when(context.getRequestProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() + .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( + ContainerProtos.WriteChunkRequestProto.newBuilder().setData(ByteString.copyFromUtf8(CONTAINER_DATA)) + .setBlockID( + ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(containerId) + .setLocalID(localId).build()).build()) + .setContainerID(containerId) + .setDatanodeUuid(UUID.randomUUID().toString()).build()); + } + + private void assertResults(boolean failWithException, AtomicReference throwable) { + if (failWithException) { + assertInstanceOf(RuntimeException.class, throwable.get()); + } else { + assertInstanceOf(StorageContainerException.class, throwable.get()); + StorageContainerException sce = (StorageContainerException) throwable.get(); + assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, sce.getResult()); + } + } + + private void setUpLogProtoReturn(ContainerStateMachine.Context context, int containerId, int localId) { + when(context.getLogProto()).thenReturn(ContainerProtos.ContainerCommandRequestProto.newBuilder() + .setCmdType(ContainerProtos.Type.WriteChunk).setWriteChunk( + ContainerProtos.WriteChunkRequestProto.newBuilder().setBlockID( + ContainerProtos.DatanodeBlockID.newBuilder(). + setContainerID(containerId).setLocalID(localId).build()).build()) + .setContainerID(containerId) + .setDatanodeUuid(UUID.randomUUID().toString()).build()); + } + + private static class ThrowableCatcher { + + private final AtomicReference caught = new AtomicReference<>(null); + + public Function asSetter() { + return t -> { + caught.set(t); + return null; + }; + } + + public AtomicReference getCaught() { + return caught; + } + + public Throwable getReceived() { + return caught.get(); + } + + public void reset() { + caught.set(null); + } + } } From 6bb5098cad1f4e2e255da93083d2bb3762bf7602 Mon Sep 17 00:00:00 2001 From: Chia-Chuan Yu Date: Thu, 20 Mar 2025 17:50:26 +0800 Subject: [PATCH 08/10] Fixed styles --- .../transport/server/ratis/TestContainerStateMachine.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java index cf264b26f85b..2b4f572854bf 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java @@ -78,7 +78,7 @@ abstract class TestContainerStateMachine { .setNameFormat("ChunkWriter-" + i + "-%d") .build())).collect(Collectors.toList()); private final boolean isLeader; - private final String CONTAINER_DATA = "Test Data"; + private static final String CONTAINER_DATA = "Test Data"; TestContainerStateMachine(boolean isLeader) { this.isLeader = isLeader; From 804c5c1665aabd8bf69c47018eafea0c5862fb11 Mon Sep 17 00:00:00 2001 From: Chia-Chuan Yu Date: Thu, 20 Mar 2025 18:09:18 +0800 Subject: [PATCH 09/10] Fixed findbugs --- .../transport/server/ratis/TestContainerStateMachine.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java index 2b4f572854bf..703e04793517 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java @@ -51,6 +51,7 @@ import org.apache.ratis.proto.RaftProtos; import org.apache.ratis.protocol.Message; import org.apache.ratis.protocol.RaftGroup; +import org.apache.ratis.protocol.RaftGroupId; import org.apache.ratis.protocol.RaftPeer; import org.apache.ratis.server.DivisionInfo; import org.apache.ratis.server.RaftServer; @@ -101,6 +102,8 @@ public void setup() throws IOException { when(division.getInfo()).thenReturn(info); when(info.isLeader()).thenReturn(isLeader); when(ratisServer.getServerDivision(any())).thenReturn(division); + stateMachine = new ContainerStateMachine(null, + RaftGroupId.randomId(), dispatcher, controller, executor, ratisServer, conf, "containerOp"); } From 20b6e928a15a7c436becefe2fb422fe777997706 Mon Sep 17 00:00:00 2001 From: Chia-Chuan Yu Date: Sat, 22 Mar 2025 21:21:39 +0800 Subject: [PATCH 10/10] Applied comments --- .../transport/server/ratis/TestContainerStateMachine.java | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java index 703e04793517..f021ca3ffd6d 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestContainerStateMachine.java @@ -17,6 +17,7 @@ package org.apache.hadoop.ozone.container.common.transport.server.ratis; +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.HDDS_CONTAINER_RATIS_STATEMACHINE_WRITE_WAIT_INTERVAL; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -31,7 +32,6 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder; import java.io.IOException; -import java.lang.reflect.Field; import java.util.List; import java.util.UUID; import java.util.concurrent.CompletableFuture; @@ -87,6 +87,8 @@ abstract class TestContainerStateMachine { @BeforeEach public void setup() throws IOException { + conf.setTimeDuration(HDDS_CONTAINER_RATIS_STATEMACHINE_WRITE_WAIT_INTERVAL, + 1000_000_000, TimeUnit.NANOSECONDS); dispatcher = mock(ContainerDispatcher.class); ContainerController controller = mock(ContainerController.class); XceiverServerRatis ratisServer = mock(XceiverServerRatis.class); @@ -226,9 +228,6 @@ public void testWriteTimout() throws Exception { setUpMockRequestProtoReturn(context, 1, 1); ThrowableCatcher catcher = new ThrowableCatcher(); - Field writeChunkWaitMaxNs = stateMachine.getClass().getDeclaredField("writeChunkWaitMaxNs"); - writeChunkWaitMaxNs.setAccessible(true); - writeChunkWaitMaxNs.set(stateMachine, 1000_000_000); CompletableFuture firstWrite = stateMachine.write(entry, trx); Thread.sleep(2000); CompletableFuture secondWrite = stateMachine.write(entryNext, trx);