chihsuan commented on code in PR #11029:
URL: https://github.com/apache/ozone/pull/11029#discussion_r3794542073
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientMetrics.java:
##########
@@ -105,54 +105,44 @@ public void testMetrics(@TempDir Path metaDir) throws
Exception {
assertCounter("CreateContainerLatencyNumOps", 1L, containerMetrics);
breakFlag = false;
- latch = new CountDownLatch(1);
-
- int numRequest = 10;
- List<CompletableFuture<ContainerCommandResponseProto>> computeResults
- = new ArrayList<>();
- // start new thread to send async requests
- Thread sendThread = new Thread(() -> {
- while (!breakFlag) {
+ int numSenderThreads = 10;
+ latch = new CountDownLatch(numSenderThreads);
+ List<CompletableFuture<ContainerCommandResponseProto>> computeResults =
+ Collections.synchronizedList(new ArrayList<>());
+ XceiverClientMetrics clientMetrics =
+ XceiverClientManager.getXceiverClientMetrics();
+
+ for (int i = 0; i < numSenderThreads; i++) {
+ Thread sendThread = new Thread(() -> {
try {
- // use async interface for testing pending metrics
- for (int i = 0; i < numRequest; i++) {
- BlockID blockID = ContainerTestHelper.
-
getTestBlockID(container.getContainerInfo().getContainerID());
- ContainerProtos.ContainerCommandRequestProto smallFileRequest;
-
- smallFileRequest = ContainerTestHelper.getWriteSmallFileRequest(
- client.getPipeline(), blockID, 1024);
- CompletableFuture<ContainerProtos.ContainerCommandResponseProto>
- response =
- client.sendCommandAsync(smallFileRequest).getResponse();
- computeResults.add(response);
+ while (!breakFlag) {
+ BlockID blockID = ContainerTestHelper.getTestBlockID(
+ container.getContainerInfo().getContainerID());
+ ContainerCommandRequestProto smallFileRequest =
+ ContainerTestHelper.getWriteSmallFileRequest(
+ client.getPipeline(), blockID, 1024);
+ computeResults.add(
+ client.sendCommandAsync(smallFileRequest).getResponse());
}
-
- Thread.sleep(1000);
} catch (Exception ignored) {
Review Comment:
Since we are already touching this block, would it make sense to keep the
first failure somewhere and surface it if the wait times out? Right now,
nothing is logged.
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientMetrics.java:
##########
@@ -105,54 +105,44 @@ public void testMetrics(@TempDir Path metaDir) throws
Exception {
assertCounter("CreateContainerLatencyNumOps", 1L, containerMetrics);
breakFlag = false;
- latch = new CountDownLatch(1);
-
- int numRequest = 10;
- List<CompletableFuture<ContainerCommandResponseProto>> computeResults
- = new ArrayList<>();
- // start new thread to send async requests
- Thread sendThread = new Thread(() -> {
- while (!breakFlag) {
+ int numSenderThreads = 10;
+ latch = new CountDownLatch(numSenderThreads);
+ List<CompletableFuture<ContainerCommandResponseProto>> computeResults =
+ Collections.synchronizedList(new ArrayList<>());
+ XceiverClientMetrics clientMetrics =
+ XceiverClientManager.getXceiverClientMetrics();
+
+ for (int i = 0; i < numSenderThreads; i++) {
+ Thread sendThread = new Thread(() -> {
try {
Review Comment:
I wonder if we should keep the try inside the loop? With the current
structure, the first transient error permanently stops the sender, whereas the
previous code retried.
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientMetrics.java:
##########
@@ -105,54 +105,44 @@ public void testMetrics(@TempDir Path metaDir) throws
Exception {
assertCounter("CreateContainerLatencyNumOps", 1L, containerMetrics);
breakFlag = false;
- latch = new CountDownLatch(1);
-
- int numRequest = 10;
- List<CompletableFuture<ContainerCommandResponseProto>> computeResults
- = new ArrayList<>();
- // start new thread to send async requests
- Thread sendThread = new Thread(() -> {
- while (!breakFlag) {
+ int numSenderThreads = 10;
+ latch = new CountDownLatch(numSenderThreads);
+ List<CompletableFuture<ContainerCommandResponseProto>> computeResults =
+ Collections.synchronizedList(new ArrayList<>());
+ XceiverClientMetrics clientMetrics =
+ XceiverClientManager.getXceiverClientMetrics();
+
+ for (int i = 0; i < numSenderThreads; i++) {
+ Thread sendThread = new Thread(() -> {
try {
- // use async interface for testing pending metrics
- for (int i = 0; i < numRequest; i++) {
- BlockID blockID = ContainerTestHelper.
-
getTestBlockID(container.getContainerInfo().getContainerID());
- ContainerProtos.ContainerCommandRequestProto smallFileRequest;
-
- smallFileRequest = ContainerTestHelper.getWriteSmallFileRequest(
- client.getPipeline(), blockID, 1024);
- CompletableFuture<ContainerProtos.ContainerCommandResponseProto>
- response =
- client.sendCommandAsync(smallFileRequest).getResponse();
- computeResults.add(response);
+ while (!breakFlag) {
+ BlockID blockID = ContainerTestHelper.getTestBlockID(
+ container.getContainerInfo().getContainerID());
+ ContainerCommandRequestProto smallFileRequest =
+ ContainerTestHelper.getWriteSmallFileRequest(
+ client.getPipeline(), blockID, 1024);
+ computeResults.add(
+ client.sendCommandAsync(smallFileRequest).getResponse());
}
-
- Thread.sleep(1000);
} catch (Exception ignored) {
+ } finally {
+ latch.countDown();
}
- }
-
- latch.countDown();
- });
- sendThread.start();
+ });
+ sendThread.start();
+ }
GenericTestUtils.waitFor(() -> {
// check if pending metric count is increased
- MetricsRecordBuilder metric =
- getMetrics(XceiverClientMetrics.SOURCE_NAME);
- long pendingOps = getLongCounter("PendingOps", metric);
- long pendingPutSmallFileOps =
- getLongCounter("numPendingPutSmallFile", metric);
-
- if (pendingOps > 0 && pendingPutSmallFileOps > 0) {
+ if (clientMetrics.getPendingContainerOpCountMetrics(
Review Comment:
The old predicate required both `pendingOps > 0 and numPendingPutSmallFile >
0` but now only the per-type counter is checked. So a regression that stops
incrementing the aggregate while still incrementing the per-type counter would
pass. Should we keep both?
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientMetrics.java:
##########
@@ -167,6 +157,9 @@ public void testMetrics(@TempDir Path metaDir) throws
Exception {
return true;
}, 100, 60000);
+ GenericTestUtils.waitFor(() ->
+ clientMetrics.getPendingContainerOpCountMetrics(
+ ContainerProtos.Type.PutSmallFile) == 0, 10, 5000);
Review Comment:
What do you think about waiting for both `PendingOps` and
`numPendingPutSmallFile` to reach zero? We assert both below, and this avoids
relying on their internal decrement order.
--
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]
