[FLINK-3535] [runtime-web] Decrease log verbosity of StackTraceSampleCoordinator

This closes #1732.

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTracker.java

@@ -33,6 +33,7 @@ import scala.concurrent.ExecutionContext;
 import scala.concurrent.Future;
 import scala.concurrent.duration.FiniteDuration;
 
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -170,6 +171,10 @@ public class BackPressureStatsTracker {
 				if (executionContext != null) {
 					pendingStats.add(vertex);
 
+					if (LOG.isDebugEnabled()) {
+						LOG.debug("Triggering stack trace sample for tasks: " + Arrays.toString(vertex.getTaskVertices()));
+					}
+
 					Future<StackTraceSample> sample = coordinator.triggerStackTraceSample(
 							vertex.getTaskVertices(),
 							numSamples,
@@ -246,7 +251,7 @@ public class BackPressureStatsTracker {
 						OperatorBackPressureStats stats = createStatsFromSample(success);
 						operatorStatsCache.put(vertex, stats);
 					} else {
-						LOG.warn("Failed to gather stack trace sample.", failure);
+						LOG.debug("Failed to gather stack trace sample.", failure);
 					}
 				} catch (Throwable t) {
 					LOG.error("Error during stats completion.", t);
@@ -278,7 +283,7 @@ public class BackPressureStatsTracker {
 				if (sampledTasks.contains(taskId)) {
 					subtaskIndexMap.put(taskId, task.getParallelSubtaskIndex());
 				} else {
-					throw new RuntimeException("Outdated sample. A task, which is part of the " +
+					LOG.debug("Outdated sample. A task, which is part of the " +
 							"sample has been reset.");
 				}
 			}

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/StackTraceSampleCoordinator.java

@@ -179,7 +179,9 @@ public class StackTraceSampleCoordinator {
 										pending.getSampleId());
 
 								pending.discard(new RuntimeException("Time out"));
-								pendingSamples.remove(pending.getSampleId());
+								if (pendingSamples.remove(pending.getSampleId()) != null) {
+									rememberRecentSampleId(pending.getSampleId());
+								}
 							}
 						}
 					} catch (Throwable t) {
@@ -319,7 +321,9 @@ public class StackTraceSampleCoordinator {
 							sampleId, executionId);
 				}
 			} else {
-				throw new IllegalStateException("Unknown sample ID " + sampleId);
+				if (LOG.isDebugEnabled()) {
+					LOG.debug("Unknown sample ID " + sampleId);
+				}
 			}
 		}
 	}

flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/StackTraceSampleCoordinatorTest.java

@@ -226,17 +226,13 @@ public class StackTraceSampleCoordinatorTest {
 		Throwable cause = sampleFuture.failed().value().get().get();
 		assertTrue(cause.getCause().getMessage().contains("Time out"));
 
-		// Collect after the timeout
-		try {
+		// Collect after the timeout (should be ignored)
 		ExecutionAttemptID executionId = vertices[0].getCurrentExecutionAttempt().getAttemptId();
 		coord.collectStackTraces(0, executionId, new ArrayList<StackTraceElement[]>());
-			fail("Did not throw expected Exception");
-		} catch (IllegalStateException ignored) {
-		}
 	}
 
-	/** Tests that collecting an unknown sample fails. */
-	@Test(expected = IllegalStateException.class)
+	/** Tests that collecting an unknown sample is ignored. */
+	@Test
 	public void testCollectStackTraceForUnknownSample() throws Exception {
 		coord.collectStackTraces(0, new ExecutionAttemptID(), new ArrayList<StackTraceElement[]>());
 	}

flink-runtime/src/main/scala/org/apache/flink/runtime/messages/StackTraceSampleMessages.scala

@@ -63,7 +63,11 @@ object StackTraceSampleMessages {
       sampleId: Int,
       executionId: ExecutionAttemptID,
       samples: java.util.List[Array[StackTraceElement]])
-    extends StackTraceSampleMessages
+    extends StackTraceSampleMessages {
+
+    override def toString: String =
+      s"ResponseStackTraceSampleSuccess($sampleId, $executionId, ${samples.size()} samples)"
+  }
 
   /**
     * Response after a failed stack trace sample (sent by the task managers to