You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2020/05/20 13:29:17 UTC
[flink] branch release-1.11 updated: [FLINK-17780][checkpointing]
Add task name to log statements of ChannelStateWriter.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new 90ece8c [FLINK-17780][checkpointing] Add task name to log statements of ChannelStateWriter.
90ece8c is described below
commit 90ece8c119ca1f748db324a301ac82a44ecda185
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Sun May 17 22:59:03 2020 +0200
[FLINK-17780][checkpointing] Add task name to log statements of ChannelStateWriter.
Add task name to the executor thread and to all method of ChannelStateWriter (as they can be called from any other thread), such that log statements can be connected to the respective task.
---
.../ChannelStateWriteRequestExecutorImpl.java | 21 +++++----
.../checkpoint/channel/ChannelStateWriterImpl.java | 51 ++++++++++++++--------
.../ChannelStateWriteRequestExecutorImplTest.java | 14 +++---
.../channel/ChannelStateWriterImplTest.java | 15 ++++---
.../runtime/state/ChannelPersistenceITCase.java | 2 +-
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 2 +-
6 files changed, 67 insertions(+), 38 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequestExecutorImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequestExecutorImpl.java
index e87a21c..3ad8982 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequestExecutorImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequestExecutorImpl.java
@@ -51,15 +51,20 @@ class ChannelStateWriteRequestExecutorImpl implements ChannelStateWriteRequestEx
private final Thread thread;
private volatile Exception thrown = null;
private volatile boolean wasClosed = false;
+ private final String taskName;
- ChannelStateWriteRequestExecutorImpl(ChannelStateWriteRequestDispatcher dispatcher) {
- this(dispatcher, new LinkedBlockingDeque<>(DEFAULT_HANDOVER_CAPACITY));
+ ChannelStateWriteRequestExecutorImpl(String taskName, ChannelStateWriteRequestDispatcher dispatcher) {
+ this(taskName, dispatcher, new LinkedBlockingDeque<>(DEFAULT_HANDOVER_CAPACITY));
}
- ChannelStateWriteRequestExecutorImpl(ChannelStateWriteRequestDispatcher dispatcher, BlockingDeque<ChannelStateWriteRequest> deque) {
+ ChannelStateWriteRequestExecutorImpl(
+ String taskName,
+ ChannelStateWriteRequestDispatcher dispatcher,
+ BlockingDeque<ChannelStateWriteRequest> deque) {
+ this.taskName = taskName;
this.dispatcher = dispatcher;
this.deque = deque;
- this.thread = new Thread(this::run);
+ this.thread = new Thread(this::run, "Channel state writer " + taskName);
this.thread.setDaemon(true);
}
@@ -80,7 +85,7 @@ class ChannelStateWriteRequestExecutorImpl implements ChannelStateWriteRequestEx
thrown = ExceptionUtils.firstOrSuppressed(e, thrown);
}
}
- LOG.debug("loop terminated");
+ LOG.debug("{} loop terminated", taskName);
}
private void loop() throws Exception {
@@ -89,7 +94,7 @@ class ChannelStateWriteRequestExecutorImpl implements ChannelStateWriteRequestEx
dispatcher.dispatch(deque.take());
} catch (InterruptedException e) {
if (!wasClosed) {
- LOG.debug("interrupted while waiting for a request (continue waiting)", e);
+ LOG.debug(taskName + " interrupted while waiting for a request (continue waiting)", e);
} else {
Thread.currentThread().interrupt();
}
@@ -101,7 +106,7 @@ class ChannelStateWriteRequestExecutorImpl implements ChannelStateWriteRequestEx
Throwable cause = thrown == null ? new CancellationException() : thrown;
List<ChannelStateWriteRequest> drained = new ArrayList<>();
deque.drainTo(drained);
- LOG.info("discarding {} drained requests", drained.size());
+ LOG.info("{} discarding {} drained requests", taskName, drained.size());
closeAll(drained.stream().<AutoCloseable>map(request -> () -> request.cancel(cause)).collect(Collectors.toList()));
}
@@ -150,7 +155,7 @@ class ChannelStateWriteRequestExecutorImpl implements ChannelStateWriteRequestEx
if (!thread.isAlive()) {
Thread.currentThread().interrupt();
}
- LOG.debug("interrupted while waiting for the writer thread to die", e);
+ LOG.debug(taskName + " interrupted while waiting for the writer thread to die", e);
}
}
if (thrown != null) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index b6fa588..3f56b15 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -58,6 +58,7 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
private static final Logger LOG = LoggerFactory.getLogger(ChannelStateWriterImpl.class);
private static final int DEFAULT_MAX_CHECKPOINTS = 5; // currently, only single in-flight checkpoint is supported
+ private final String taskName;
private final ChannelStateWriteRequestExecutor executor;
private final ConcurrentMap<Long, ChannelStateWriteResult> results;
private final int maxCheckpoints;
@@ -65,26 +66,31 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
/**
* Creates a {@link ChannelStateWriterImpl} with {@link #DEFAULT_MAX_CHECKPOINTS} as {@link #maxCheckpoints}.
*/
- public ChannelStateWriterImpl(CheckpointStorageWorkerView streamFactoryResolver) {
- this(streamFactoryResolver, DEFAULT_MAX_CHECKPOINTS);
+ public ChannelStateWriterImpl(String taskName, CheckpointStorageWorkerView streamFactoryResolver) {
+ this(taskName, streamFactoryResolver, DEFAULT_MAX_CHECKPOINTS);
}
/**
* Creates a {@link ChannelStateWriterImpl} with {@link ChannelStateSerializerImpl default} {@link ChannelStateSerializer},
* and a {@link ChannelStateWriteRequestExecutorImpl}.
- *
- * @param maxCheckpoints maximum number of checkpoints to be written currently or finished but not taken yet.
+ * @param taskName
* @param streamFactoryResolver a factory to obtain output stream factory for a given checkpoint
+ * @param maxCheckpoints maximum number of checkpoints to be written currently or finished but not taken yet.
*/
- ChannelStateWriterImpl(CheckpointStorageWorkerView streamFactoryResolver, int maxCheckpoints) {
+ ChannelStateWriterImpl(String taskName, CheckpointStorageWorkerView streamFactoryResolver, int maxCheckpoints) {
this(
+ taskName,
new ConcurrentHashMap<>(maxCheckpoints),
- new ChannelStateWriteRequestExecutorImpl(new ChannelStateWriteRequestDispatcherImpl(streamFactoryResolver, new ChannelStateSerializerImpl())),
- maxCheckpoints
- );
+ new ChannelStateWriteRequestExecutorImpl(taskName, new ChannelStateWriteRequestDispatcherImpl(streamFactoryResolver, new ChannelStateSerializerImpl())),
+ maxCheckpoints);
}
- ChannelStateWriterImpl(ConcurrentMap<Long, ChannelStateWriteResult> results, ChannelStateWriteRequestExecutor executor, int maxCheckpoints) {
+ ChannelStateWriterImpl(
+ String taskName,
+ ConcurrentMap<Long, ChannelStateWriteResult> results,
+ ChannelStateWriteRequestExecutor executor,
+ int maxCheckpoints) {
+ this.taskName = taskName;
this.results = results;
this.maxCheckpoints = maxCheckpoints;
this.executor = executor;
@@ -93,7 +99,7 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
@Override
public void start(long checkpointId, CheckpointOptions checkpointOptions) {
results.keySet().forEach(oldCheckpointId -> abort(oldCheckpointId, new Exception("Starting new checkpoint " + checkpointId)));
- LOG.debug("start checkpoint {} ({})", checkpointId, checkpointOptions);
+ LOG.debug("{} starting checkpoint {} ({})", taskName, checkpointId, checkpointOptions);
ChannelStateWriteResult result = new ChannelStateWriteResult();
ChannelStateWriteResult put = results.computeIfAbsent(checkpointId, id -> {
Preconditions.checkState(results.size() < maxCheckpoints, "results.size() > maxCheckpoints", results.size(), maxCheckpoints);
@@ -105,32 +111,42 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
@Override
public void addInputData(long checkpointId, InputChannelInfo info, int startSeqNum, CloseableIterator<Buffer> iterator) {
- LOG.debug("add input data, checkpoint id: {}, channel: {}, startSeqNum: {}", checkpointId, info, startSeqNum);
+ LOG.debug(
+ "{} adding input data, checkpoint id: {}, channel: {}, startSeqNum: {}",
+ taskName,
+ checkpointId,
+ info,
+ startSeqNum);
enqueue(write(checkpointId, info, iterator), false);
}
@Override
public void addOutputData(long checkpointId, ResultSubpartitionInfo info, int startSeqNum, Buffer... data) {
- LOG.debug("add output data, checkpoint id: {}, channel: {}, startSeqNum: {}, num buffers: {}",
- checkpointId, info, startSeqNum, data == null ? 0 : data.length);
+ LOG.debug(
+ "{} adding output data, checkpoint id: {}, channel: {}, startSeqNum: {}, num buffers: {}",
+ taskName,
+ checkpointId,
+ info,
+ startSeqNum,
+ data == null ? 0 : data.length);
enqueue(write(checkpointId, info, checkBufferType(data)), false);
}
@Override
public void finishInput(long checkpointId) {
- LOG.debug("finish input data, checkpoint id: {}", checkpointId);
+ LOG.debug("{} finishing input data, checkpoint id: {}", taskName, checkpointId);
enqueue(completeInput(checkpointId), false);
}
@Override
public void finishOutput(long checkpointId) {
- LOG.debug("finish output data, checkpoint id: {}", checkpointId);
+ LOG.debug("{} finishing output data, checkpoint id: {}", taskName, checkpointId);
enqueue(completeOutput(checkpointId), false);
}
@Override
public void abort(long checkpointId, Throwable cause) {
- LOG.debug("abort, checkpoint id: {}", checkpointId);
+ LOG.debug("{} aborting, checkpoint id: {}", taskName, checkpointId);
enqueue(ChannelStateWriteRequest.abort(checkpointId, cause), true); // abort already started
enqueue(ChannelStateWriteRequest.abort(checkpointId, cause), false); // abort enqueued but not started
results.remove(checkpointId);
@@ -138,7 +154,7 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
@Override
public ChannelStateWriteResult getWriteResult(long checkpointId) {
- LOG.debug("requested write result, checkpoint id: {}", checkpointId);
+ LOG.debug("{} requested write result, checkpoint id: {}", taskName, checkpointId);
ChannelStateWriteResult result = results.get(checkpointId);
Preconditions.checkArgument(result != null, "channel state write result not found for checkpoint id " + checkpointId);
return result;
@@ -146,6 +162,7 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
@Override
public void stop(long checkpointId) {
+ LOG.debug("{} stopping checkpoint id: {}", taskName, checkpointId);
results.remove(checkpointId);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequestExecutorImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequestExecutorImplTest.java
index a299b34..431d967 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequestExecutorImplTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequestExecutorImplTest.java
@@ -39,6 +39,8 @@ import static org.junit.Assert.fail;
*/
public class ChannelStateWriteRequestExecutorImplTest {
+ private static final String TASK_NAME = "test task";
+
@Test(expected = IllegalStateException.class)
public void testCloseAfterSubmit() throws Exception {
testCloseAfterSubmit(ChannelStateWriteRequestExecutor::submit);
@@ -61,7 +63,7 @@ public class ChannelStateWriteRequestExecutorImplTest {
private void testCloseAfterSubmit(BiConsumerWithException<ChannelStateWriteRequestExecutor, ChannelStateWriteRequest, Exception> requestFun) throws Exception {
WorkerClosingDeque closingDeque = new WorkerClosingDeque();
- ChannelStateWriteRequestExecutorImpl worker = new ChannelStateWriteRequestExecutorImpl(NO_OP, closingDeque);
+ ChannelStateWriteRequestExecutorImpl worker = new ChannelStateWriteRequestExecutorImpl(TASK_NAME, NO_OP, closingDeque);
closingDeque.setWorker(worker);
TestWriteRequest request = new TestWriteRequest();
requestFun.accept(worker, request);
@@ -73,7 +75,7 @@ public class ChannelStateWriteRequestExecutorImplTest {
TestWriteRequest request = new TestWriteRequest();
LinkedBlockingDeque<ChannelStateWriteRequest> deque = new LinkedBlockingDeque<>();
try {
- submitAction.accept(new ChannelStateWriteRequestExecutorImpl(NO_OP, deque), request);
+ submitAction.accept(new ChannelStateWriteRequestExecutorImpl(TASK_NAME, NO_OP, deque), request);
} catch (IllegalStateException e) {
// expected: executor not started;
return;
@@ -91,7 +93,7 @@ public class ChannelStateWriteRequestExecutorImplTest {
LinkedBlockingDeque<ChannelStateWriteRequest> deque = new LinkedBlockingDeque<>();
deque.add(request);
TestRequestDispatcher requestProcessor = new TestRequestDispatcher();
- ChannelStateWriteRequestExecutorImpl worker = new ChannelStateWriteRequestExecutorImpl(requestProcessor, deque);
+ ChannelStateWriteRequestExecutorImpl worker = new ChannelStateWriteRequestExecutorImpl(TASK_NAME, requestProcessor, deque);
worker.close();
worker.run();
@@ -105,7 +107,7 @@ public class ChannelStateWriteRequestExecutorImplTest {
public void testIgnoresInterruptsWhileRunning() throws Exception {
TestRequestDispatcher requestProcessor = new TestRequestDispatcher();
LinkedBlockingDeque<ChannelStateWriteRequest> deque = new LinkedBlockingDeque<>();
- try (ChannelStateWriteRequestExecutorImpl worker = new ChannelStateWriteRequestExecutorImpl(requestProcessor, deque)) {
+ try (ChannelStateWriteRequestExecutorImpl worker = new ChannelStateWriteRequestExecutorImpl(TASK_NAME, requestProcessor, deque)) {
worker.start();
worker.getThread().interrupt();
worker.submit(new TestWriteRequest());
@@ -119,7 +121,7 @@ public class ChannelStateWriteRequestExecutorImplTest {
@Test
public void testCanBeClosed() throws IOException {
TestRequestDispatcher requestProcessor = new TestRequestDispatcher();
- try (ChannelStateWriteRequestExecutorImpl worker = new ChannelStateWriteRequestExecutorImpl(requestProcessor)) {
+ try (ChannelStateWriteRequestExecutorImpl worker = new ChannelStateWriteRequestExecutorImpl(TASK_NAME, requestProcessor)) {
worker.start();
}
}
@@ -134,7 +136,7 @@ public class ChannelStateWriteRequestExecutorImplTest {
}
};
LinkedBlockingDeque<ChannelStateWriteRequest> deque = new LinkedBlockingDeque<>(Arrays.asList(new TestWriteRequest()));
- ChannelStateWriteRequestExecutorImpl worker = new ChannelStateWriteRequestExecutorImpl(throwingRequestProcessor, deque);
+ ChannelStateWriteRequestExecutorImpl worker = new ChannelStateWriteRequestExecutorImpl(TASK_NAME, throwingRequestProcessor, deque);
worker.run();
try {
worker.close();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
index 92a7e88..8c7d7f2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
@@ -46,6 +46,7 @@ import static org.junit.Assert.assertTrue;
*/
public class ChannelStateWriterImplTest {
private static final long CHECKPOINT_ID = 42L;
+ private static final String TASK_NAME = "test";
@Test(expected = IllegalArgumentException.class)
public void testAddEventBuffer() throws Exception {
@@ -122,7 +123,11 @@ public class ChannelStateWriterImplTest {
public void testBuffersRecycledOnError() throws Exception {
unwrappingError(TestException.class, () -> {
NetworkBuffer buffer = getBuffer();
- try (ChannelStateWriterImpl writer = new ChannelStateWriterImpl(new ConcurrentHashMap<>(), failingWorker(), 5)) {
+ try (ChannelStateWriterImpl writer = new ChannelStateWriterImpl(
+ TASK_NAME,
+ new ConcurrentHashMap<>(),
+ failingWorker(),
+ 5)) {
writer.open();
callAddInputData(writer, buffer);
} finally {
@@ -179,7 +184,7 @@ public class ChannelStateWriterImplTest {
@Test(expected = TestException.class)
public void testRethrowOnNextCall() throws Exception {
SyncChannelStateWriteRequestExecutor worker = new SyncChannelStateWriteRequestExecutor();
- ChannelStateWriterImpl writer = new ChannelStateWriterImpl(new ConcurrentHashMap<>(), worker, 5);
+ ChannelStateWriterImpl writer = new ChannelStateWriterImpl(TASK_NAME, new ConcurrentHashMap<>(), worker, 5);
writer.open();
worker.setThrown(new TestException());
unwrappingError(TestException.class, () -> callStart(writer));
@@ -203,7 +208,7 @@ public class ChannelStateWriterImplTest {
@Test(expected = IllegalStateException.class)
public void testStartNotOpened() throws Exception {
unwrappingError(IllegalStateException.class, () -> {
- try (ChannelStateWriterImpl writer = new ChannelStateWriterImpl(getStreamFactoryFactory())) {
+ try (ChannelStateWriterImpl writer = new ChannelStateWriterImpl(TASK_NAME, getStreamFactoryFactory())) {
callStart(writer);
}
});
@@ -269,7 +274,7 @@ public class ChannelStateWriterImplTest {
private void runWithSyncWorker(BiConsumerWithException<ChannelStateWriter, SyncChannelStateWriteRequestExecutor, Exception> testFn) throws Exception {
try (
SyncChannelStateWriteRequestExecutor worker = new SyncChannelStateWriteRequestExecutor();
- ChannelStateWriterImpl writer = new ChannelStateWriterImpl(new ConcurrentHashMap<>(), worker, 5)
+ ChannelStateWriterImpl writer = new ChannelStateWriterImpl(TASK_NAME, new ConcurrentHashMap<>(), worker, 5)
) {
writer.open();
testFn.accept(writer, worker);
@@ -278,7 +283,7 @@ public class ChannelStateWriterImplTest {
}
private ChannelStateWriterImpl openWriter() {
- ChannelStateWriterImpl writer = new ChannelStateWriterImpl(getStreamFactoryFactory());
+ ChannelStateWriterImpl writer = new ChannelStateWriterImpl(TASK_NAME, getStreamFactoryFactory());
writer.open();
return writer;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChannelPersistenceITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChannelPersistenceITCase.java
index a77dbbf..b696c77 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChannelPersistenceITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChannelPersistenceITCase.java
@@ -99,7 +99,7 @@ public class ChannelPersistenceITCase {
int maxStateSize = sizeOfBytes(icMap) + sizeOfBytes(rsMap) + Long.BYTES * 2;
Map<InputChannelInfo, Buffer> icBuffers = wrapWithBuffers(icMap);
Map<ResultSubpartitionInfo, Buffer> rsBuffers = wrapWithBuffers(rsMap);
- try (ChannelStateWriterImpl writer = new ChannelStateWriterImpl(getStreamFactoryFactory(maxStateSize))) {
+ try (ChannelStateWriterImpl writer = new ChannelStateWriterImpl("test", getStreamFactoryFactory(maxStateSize))) {
writer.open();
writer.start(checkpointId, new CheckpointOptions(CHECKPOINT, new CheckpointStorageLocationReference("poly".getBytes())));
for (Map.Entry<InputChannelInfo, Buffer> e : icBuffers.entrySet()) {
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
index 7508a16..9fc7927 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
@@ -152,7 +152,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
}
private ChannelStateWriter openChannelStateWriter() {
- ChannelStateWriterImpl writer = new ChannelStateWriterImpl(this.checkpointStorage);
+ ChannelStateWriterImpl writer = new ChannelStateWriterImpl(taskName, checkpointStorage);
writer.open();
return writer;
}