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_r344603414
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java
##########
@@ -555,138 +555,88 @@ public void testFailingScheduleOrUpdateConsumers()
throws Exception {
}
//
------------------------------------------------------------------------
- // Stack trace sample
+ // Back pressure request
//
------------------------------------------------------------------------
/**
- * Tests sampling of task stack traces.
+ * Tests request of task back pressure.
*/
- @Test(timeout = 10000L)
- @SuppressWarnings("unchecked")
- public void testRequestStackTraceSample() throws Exception {
- final ExecutionAttemptID eid = new ExecutionAttemptID();
- final TaskDeploymentDescriptor tdd =
createTestTaskDeploymentDescriptor("test task", eid,
BlockingNoOpInvokable.class);
-
- final int sampleId1 = 112223;
- final int sampleId2 = 19230;
- final int sampleId3 = 1337;
- final int sampleId4 = 44;
+ @Test(timeout = 20000L)
+ public void testRequestTaskBackPressure() throws Exception {
+ final NettyShuffleDescriptor shuffleDescriptor =
createRemoteWithIdAndLocation(
+ new IntermediateResultPartitionID(),
ResourceID.generate());
+ final TaskDeploymentDescriptor tdd =
createSender(shuffleDescriptor, OutputBlockedInvokable.class);
+ final ExecutionAttemptID executionAttemptID =
tdd.getExecutionAttemptId();
final CompletableFuture<Void> taskRunningFuture = new
CompletableFuture<>();
final CompletableFuture<Void> taskCanceledFuture = new
CompletableFuture<>();
- try (TaskSubmissionTestEnvironment env =
- new TaskSubmissionTestEnvironment.Builder(jobId)
- .setSlotSize(1)
- .addTaskManagerActionListener(eid,
ExecutionState.RUNNING, taskRunningFuture)
- .addTaskManagerActionListener(eid,
ExecutionState.CANCELED, taskCanceledFuture)
- .build()) {
- TaskExecutorGateway tmGateway =
env.getTaskExecutorGateway();
- TaskSlotTable taskSlotTable = env.getTaskSlotTable();
+ final Configuration configuration = new Configuration();
+ configuration.set(WebOptions.BACKPRESSURE_NUM_SAMPLES, 10);
+ configuration.set(WebOptions.BACKPRESSURE_DELAY, 200);
Review comment:
The default delay 50 does not make sense?
----------------------------------------------------------------
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
