zhijiangW commented on a change in pull request #10083:
[FLINK-14472][runtime]Implement back-pressure monitor with non-blocking outputs.
URL: https://github.com/apache/flink/pull/10083#discussion_r344029460
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/backpressure/BackPressureStatsTrackerImplTest.java
##########
@@ -29,155 +28,242 @@
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.util.TestLogger;
-import org.junit.Assert;
import org.junit.Test;
-import org.mockito.Matchers;
-import org.mockito.Mockito;
+import org.mockito.ArgumentMatchers;
+import org.mockito.stubbing.OngoingStubbing;
-import java.util.ArrayList;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/**
- * Tests for the BackPressureStatsTrackerImpl.
+ * Tests for the {@link BackPressureStatsTrackerImpl}.
*/
public class BackPressureStatsTrackerImplTest extends TestLogger {
- /** Tests simple statistics with fake stack traces. */
@Test
- @SuppressWarnings("unchecked")
- public void testTriggerStackTraceSample() throws Exception {
- CompletableFuture<StackTraceSample> sampleFuture = new
CompletableFuture<>();
-
- StackTraceSampleCoordinator sampleCoordinator =
Mockito.mock(StackTraceSampleCoordinator.class);
- Mockito.when(sampleCoordinator.triggerStackTraceSample(
- Matchers.any(ExecutionVertex[].class),
- Matchers.anyInt(),
- Matchers.any(Time.class),
- Matchers.anyInt())).thenReturn(sampleFuture);
+ public void testGetOperatorBackPressureStats() throws Exception {
+ final ExecutionJobVertex jobVertex =
createExecutionJobVertex(JobStatus.RUNNING);
- ExecutionGraph graph = Mockito.mock(ExecutionGraph.class);
- Mockito.when(graph.getState()).thenReturn(JobStatus.RUNNING);
+ final int requestId = 0;
+ final long startTime = System.currentTimeMillis();
+ final long endTime = startTime + 1;
+ final double backPressureRatio = 0.1;
- // Same Thread execution context
- Mockito.when(graph.getFutureExecutor()).thenReturn(new
Executor() {
+ final BackPressureStats backPressureStats =
createBackPressureStats(
+ jobVertex.getTaskVertices(), requestId, startTime,
endTime, backPressureRatio);
+ final BackPressureStatsTracker tracker =
createBackPressureTracker(600000, 10000, backPressureStats);
- @Override
- public void execute(Runnable runnable) {
- runnable.run();
- }
- });
+ // trigger back pressure stats request
+ tracker.getOperatorBackPressureStats(jobVertex);
- ExecutionVertex[] taskVertices = new ExecutionVertex[4];
+ Optional<OperatorBackPressureStats> optionalStats =
tracker.getOperatorBackPressureStats(jobVertex);
+ assertTrue(optionalStats.isPresent());
+ OperatorBackPressureStats stats = optionalStats.get();
- ExecutionJobVertex jobVertex =
Mockito.mock(ExecutionJobVertex.class);
- Mockito.when(jobVertex.getJobId()).thenReturn(new JobID());
- Mockito.when(jobVertex.getJobVertexId()).thenReturn(new
JobVertexID());
- Mockito.when(jobVertex.getGraph()).thenReturn(graph);
-
Mockito.when(jobVertex.getTaskVertices()).thenReturn(taskVertices);
-
- taskVertices[0] = mockExecutionVertex(jobVertex, 0);
- taskVertices[1] = mockExecutionVertex(jobVertex, 1);
- taskVertices[2] = mockExecutionVertex(jobVertex, 2);
- taskVertices[3] = mockExecutionVertex(jobVertex, 3);
+ checkOperatorBackPressureStats(jobVertex.getTaskVertices(),
requestId, endTime, backPressureRatio, stats);
+ }
- int numSamples = 100;
- Time delayBetweenSamples = Time.milliseconds(100L);
+ @Test
+ public void testOperatorBackPressureStatsUpdate() throws Exception {
+ final ExecutionJobVertex jobVertex =
createExecutionJobVertex(JobStatus.RUNNING);
+ final int backPressureStatsRefreshInterval = 2000;
+ final long waitingTime = backPressureStatsRefreshInterval + 500;
+
+ final int requestId1 = 0;
+ final long startTime1 = System.currentTimeMillis();
+ final long endTime1 = startTime1 + 1;
+ final double backPressureRatio1 = 0.1;
+
+ final int requestId2 = 1;
+ final long startTime2 = System.currentTimeMillis() +
waitingTime;
+ final long endTime2 = startTime2 + 1;
+ final double backPressureRatio2 = 0.2;
+
+ final BackPressureStats backPressureStats1 =
createBackPressureStats(
+ jobVertex.getTaskVertices(), requestId1, startTime1,
endTime1, backPressureRatio1);
+ final BackPressureStats backPressureStats2 =
createBackPressureStats(
+ jobVertex.getTaskVertices(), requestId2, startTime2,
endTime2, backPressureRatio2);
+ final BackPressureStatsTracker tracker =
createBackPressureTracker(
+ 600000, backPressureStatsRefreshInterval,
backPressureStats1, backPressureStats2);
+
+ // trigger back pressure stats request
+
assertFalse(tracker.getOperatorBackPressureStats(jobVertex).isPresent());
+
+ Optional<OperatorBackPressureStats> optionalStats =
tracker.getOperatorBackPressureStats(jobVertex);
+ assertTrue(optionalStats.isPresent());
+ OperatorBackPressureStats stats = optionalStats.get();
+
+ final int requestId = stats.getRequestId();
+ checkOperatorBackPressureStats(jobVertex.getTaskVertices(),
requestId1, endTime1, backPressureRatio1, stats);
+
+ // should not trigger new back pressure stats request
+
assertTrue(tracker.getOperatorBackPressureStats(jobVertex).isPresent());
+ assertEquals(requestId,
tracker.getOperatorBackPressureStats(jobVertex).get().getRequestId());
+
+ // ensure that we are ready for next request
+ Thread.sleep(waitingTime);
+
+ // trigger next back pressure stats request
+
assertTrue(tracker.getOperatorBackPressureStats(jobVertex).isPresent());
+ OperatorBackPressureStats newStats =
tracker.getOperatorBackPressureStats(jobVertex).get();
+ assertNotEquals(requestId, newStats.getRequestId());
+
+ checkOperatorBackPressureStats(jobVertex.getTaskVertices(),
requestId2, endTime2, backPressureRatio2, newStats);
+ }
- BackPressureStatsTrackerImpl tracker = new
BackPressureStatsTrackerImpl(
- sampleCoordinator, 9999, numSamples,
Integer.MAX_VALUE, delayBetweenSamples);
+ @Test
+ public void testGetOperatorBackPressureStatsAfterShutDown() throws
Exception {
+ final ExecutionJobVertex jobVertex =
createExecutionJobVertex(JobStatus.RUNNING);
- // getOperatorBackPressureStats triggers stack trace sampling
-
Assert.assertFalse(tracker.getOperatorBackPressureStats(jobVertex).isPresent());
+ final int requestId = 0;
+ final long startTime = System.currentTimeMillis();
+ final long endTime = startTime + 1;
+ final double backPressureRatio = 0.1;
- Mockito.verify(sampleCoordinator,
Mockito.times(1)).triggerStackTraceSample(
- Matchers.eq(taskVertices),
- Matchers.eq(numSamples),
- Matchers.eq(delayBetweenSamples),
-
Matchers.eq(BackPressureStatsTrackerImpl.MAX_STACK_TRACE_DEPTH));
+ final BackPressureStats backPressureStats =
createBackPressureStats(
+ jobVertex.getTaskVertices(), requestId, startTime,
endTime, backPressureRatio);
+ final BackPressureStatsTracker tracker =
createBackPressureTracker(600000, 10000, backPressureStats);
- // Request back pressure stats again. This should not trigger
another sample request
-
Assert.assertTrue(!tracker.getOperatorBackPressureStats(jobVertex).isPresent());
+ tracker.shutDown();
- Mockito.verify(sampleCoordinator,
Mockito.times(1)).triggerStackTraceSample(
- Matchers.eq(taskVertices),
- Matchers.eq(numSamples),
- Matchers.eq(delayBetweenSamples),
-
Matchers.eq(BackPressureStatsTrackerImpl.MAX_STACK_TRACE_DEPTH));
+ // trigger back pressure stats request
+ tracker.getOperatorBackPressureStats(jobVertex);
-
Assert.assertTrue(!tracker.getOperatorBackPressureStats(jobVertex).isPresent());
+
assertFalse(tracker.getOperatorBackPressureStats(jobVertex).isPresent());
+ }
- // Complete the future
- Map<ExecutionAttemptID, List<StackTraceElement[]>> traces = new
HashMap<>();
- for (ExecutionVertex vertex : taskVertices) {
- List<StackTraceElement[]> taskTraces = new
ArrayList<>();
+ @Test
+ public void testStatsCleanup() throws Exception {
+ final ExecutionJobVertex jobVertex =
createExecutionJobVertex(JobStatus.RUNNING);
+ final int cleanUpInterval = 2000;
+ final long waitingTime = cleanUpInterval + 500;
- for (int i = 0; i < taskVertices.length; i++) {
- // Traces until sub task index are back
pressured
- taskTraces.add(createStackTrace(i <=
vertex.getParallelSubtaskIndex()));
- }
+ final int requestId = 0;
+ final long startTime = System.currentTimeMillis();
+ final long endTime = startTime + 1;
+ final double backPressureRatio = 0.1;
-
traces.put(vertex.getCurrentExecutionAttempt().getAttemptId(), taskTraces);
- }
+ final BackPressureStats backPressureStats =
createBackPressureStats(
+ jobVertex.getTaskVertices(), requestId, startTime,
endTime, backPressureRatio);
+ final BackPressureStatsTracker tracker =
createBackPressureTracker(2000, 10000, backPressureStats);
- int sampleId = 1231;
- int endTime = 841;
+ // trigger back pressure stats request
+ tracker.getOperatorBackPressureStats(jobVertex);
- StackTraceSample sample = new StackTraceSample(
- sampleId,
- 0,
- endTime,
- traces);
+ Optional<OperatorBackPressureStats> optionalStats =
tracker.getOperatorBackPressureStats(jobVertex);
+ assertTrue(optionalStats.isPresent());
+ OperatorBackPressureStats stats = optionalStats.get();
- // Succeed the promise
- sampleFuture.complete(sample);
+ checkOperatorBackPressureStats(jobVertex.getTaskVertices(),
requestId, endTime, backPressureRatio, stats);
-
Assert.assertTrue(tracker.getOperatorBackPressureStats(jobVertex).isPresent());
+ tracker.cleanUpOperatorStatsCache();
+
assertTrue(tracker.getOperatorBackPressureStats(jobVertex).isPresent());
- OperatorBackPressureStats stats =
tracker.getOperatorBackPressureStats(jobVertex).get();
+ // wait until we are ready to cleanup
+ Thread.sleep(waitingTime);
- // Verify the stats
- Assert.assertEquals(sampleId, stats.getSampleId());
- Assert.assertEquals(endTime, stats.getEndTimestamp());
- Assert.assertEquals(taskVertices.length,
stats.getNumberOfSubTasks());
+ tracker.cleanUpOperatorStatsCache();
+
assertFalse(tracker.getOperatorBackPressureStats(jobVertex).isPresent());
+ }
- for (int i = 0; i < taskVertices.length; i++) {
- double ratio = stats.getBackPressureRatio(i);
- // Traces until sub task index are back pressured
- Assert.assertEquals((i + 1) / ((double) 4), ratio, 0.0);
+ private void checkOperatorBackPressureStats(
+ ExecutionVertex[] taskVertices,
+ int requestId,
+ long endTime,
+ double backPressureRatio,
+ OperatorBackPressureStats stats) {
+
+ assertEquals(requestId, stats.getRequestId());
+ assertEquals(endTime, stats.getEndTimestamp());
+ assertEquals(taskVertices.length, stats.getNumberOfSubTasks());
+ for (int i = 0; i < stats.getNumberOfSubTasks(); i++) {
+ assertEquals(backPressureRatio,
stats.getBackPressureRatio(i), 0.0);
}
}
- private StackTraceElement[] createStackTrace(boolean isBackPressure) {
- if (isBackPressure) {
- return new StackTraceElement[] { new StackTraceElement(
-
BackPressureStatsTrackerImpl.EXPECTED_CLASS_NAME,
-
BackPressureStatsTrackerImpl.EXPECTED_METHOD_NAME,
- "LocalBufferPool.java",
- 133) };
- } else {
- return Thread.currentThread().getStackTrace();
- }
+ private BackPressureStatsTracker createBackPressureTracker(
+ int cleanUpInterval,
+ int backPressureStatsRefreshInterval,
+ BackPressureStats... stats) {
+
+ final BackPressureRequestCoordinator coordinator =
createBackPressureRequestCoordinator(stats);
+ return new BackPressureStatsTrackerImpl(
+ coordinator,
+ cleanUpInterval,
+ backPressureStatsRefreshInterval);
}
- private ExecutionVertex mockExecutionVertex(
- ExecutionJobVertex jobVertex,
- int subTaskIndex) {
+ private ExecutionVertex createExecutionVertex(ExecutionJobVertex
jobVertex, int subTaskIndex) {
- Execution exec = Mockito.mock(Execution.class);
- Mockito.when(exec.getAttemptId()).thenReturn(new
ExecutionAttemptID());
+ Execution exec = mock(Execution.class);
+ when(exec.getAttemptId()).thenReturn(new ExecutionAttemptID());
JobVertexID id = jobVertex.getJobVertexId();
- ExecutionVertex vertex = Mockito.mock(ExecutionVertex.class);
- Mockito.when(vertex.getJobvertexId()).thenReturn(id);
-
Mockito.when(vertex.getCurrentExecutionAttempt()).thenReturn(exec);
-
Mockito.when(vertex.getParallelSubtaskIndex()).thenReturn(subTaskIndex);
+ ExecutionVertex vertex = mock(ExecutionVertex.class);
+ when(vertex.getJobvertexId()).thenReturn(id);
+ when(vertex.getCurrentExecutionAttempt()).thenReturn(exec);
+ when(vertex.getParallelSubtaskIndex()).thenReturn(subTaskIndex);
return vertex;
}
+ private ExecutionJobVertex createExecutionJobVertex(JobStatus
jobStatus) {
+ ExecutionGraph graph = mock(ExecutionGraph.class);
+ when(graph.getState()).thenReturn(jobStatus);
+ when(graph.getFutureExecutor()).thenReturn(Runnable::run);
+
+ ExecutionVertex[] taskVertices = new ExecutionVertex[4];
+
+ ExecutionJobVertex jobVertex = mock(ExecutionJobVertex.class);
+ when(jobVertex.getJobId()).thenReturn(new JobID());
+ when(jobVertex.getJobVertexId()).thenReturn(new JobVertexID());
+ when(jobVertex.getGraph()).thenReturn(graph);
+ when(jobVertex.getTaskVertices()).thenReturn(taskVertices);
+
+ taskVertices[0] = createExecutionVertex(jobVertex, 0);
+ taskVertices[1] = createExecutionVertex(jobVertex, 1);
+ taskVertices[2] = createExecutionVertex(jobVertex, 2);
+ taskVertices[3] = createExecutionVertex(jobVertex, 3);
+
+ return jobVertex;
+ }
+
+ private BackPressureRequestCoordinator
createBackPressureRequestCoordinator(BackPressureStats... stats) {
+
+ BackPressureRequestCoordinator coordinator =
mock(BackPressureRequestCoordinator.class);
Review comment:
avoid mock
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
