This is an automated email from the ASF dual-hosted git repository.
scwhittle pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 6ddc7fec3b8 Log the System name in more places instead of the
computationId (#39665)
6ddc7fec3b8 is described below
commit 6ddc7fec3b8edbb0af1cf9c8f5af047aeb0254b7
Author: Ryan Wigglesworth <[email protected]>
AuthorDate: Fri Aug 14 11:06:28 2026 +0000
Log the System name in more places instead of the computationId (#39665)
---
.../work/processing/StreamingWorkScheduler.java | 10 +++--
.../processing/failures/WorkFailureProcessor.java | 25 ++++++-----
.../failures/WorkFailureProcessorTest.java | 49 ++++++++++++++++++----
3 files changed, 62 insertions(+), 22 deletions(-)
diff --git
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java
index 299c67128ca..27952569e0e 100644
---
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java
+++
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java
@@ -227,8 +227,9 @@ public class StreamingWorkScheduler {
ComputationState computationState, Work work,
BoundedQueueExecutorWorkHandle handle) {
Windmill.WorkItem workItem = work.getWorkItem();
String computationId = computationState.getComputationId();
- LOG.debug("Starting processing for {}:\n{}", computationId, work);
- setLoggingContextComputation(computationState.getSystemName());
+ String systemName = computationState.getSystemName();
+ LOG.debug("Starting processing for {}:\n{}", systemName, work);
+ setLoggingContextComputation(systemName);
KeyTransitionListener keyTransitionListener =
createKeyTransitionListener();
keyTransitionListener.onKeyTransition(null, work);
@@ -259,7 +260,8 @@ public class StreamingWorkScheduler {
recordProcessingStats(workBatch, workItemCommits,
executeWorkResult.stateBytesRead());
LOG.debug("Processing done for work batch size: {}", workBatch.size());
} catch (Throwable t) {
- handleProcessWorkFailure(computationState, handle.getWorkBatch(),
computationId, work, t);
+ handleProcessWorkFailure(
+ computationState, handle.getWorkBatch(), computationId, systemName,
work, t);
} finally {
List<Work> processedWorkBatch = workBatch != null ? workBatch :
ImmutableList.of(work);
// Update total processing time counters. Updating in finally clause
ensures that
@@ -453,6 +455,7 @@ public class StreamingWorkScheduler {
ComputationState computationState,
List<Work> failedBatch,
String computationId,
+ String systemName,
Work primaryWork,
Throwable t) {
try {
@@ -464,6 +467,7 @@ public class StreamingWorkScheduler {
workFailureProcessor.logAndProcessFailureBatch(
computationId,
+ systemName,
executableWorks,
t,
invalidWork ->
diff --git
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/failures/WorkFailureProcessor.java
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/failures/WorkFailureProcessor.java
index 8af1840faf9..de33d3d3961 100644
---
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/failures/WorkFailureProcessor.java
+++
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/failures/WorkFailureProcessor.java
@@ -100,6 +100,7 @@ public final class WorkFailureProcessor {
public void logAndProcessFailureBatch(
String computationId,
+ String systemName,
List<ExecutableWork> executableWorks,
Throwable t,
Consumer<Work> onInvalidWork)
@@ -107,7 +108,7 @@ public final class WorkFailureProcessor {
List<ExecutableWork> worksToRetryLocally = new java.util.ArrayList<>();
for (ExecutableWork executableWork : executableWorks) {
- switch (evaluateRetry(computationId, executableWork.work(), t)) {
+ switch (evaluateRetry(computationId, systemName, executableWork.work(),
t)) {
case DO_NOT_RETRY:
// Consider the item invalid. It will eventually be retried by
Windmill if it still needs
// to be processed.
@@ -148,12 +149,13 @@ public final class WorkFailureProcessor {
RETHROW_THROWABLE,
}
- private RetryEvaluation evaluateRetry(String computationId, Work work,
Throwable t) {
+ private RetryEvaluation evaluateRetry(
+ String computationId, String systemName, Work work, Throwable t) {
if (work.isFailed()) {
LOG.debug(
- "Execution of work for computation '{}' on sharding key '{}' failed.
"
+ "Execution of work for fused stage '{}' on sharding key '{}' failed.
"
+ "Work is already marked as failed, not retrying locally.",
- computationId,
+ systemName,
work.getWorkItem().getShardingKey());
return RetryEvaluation.DO_NOT_RETRY;
}
@@ -166,9 +168,9 @@ public final class WorkFailureProcessor {
if (isOutOfMemoryError(parsedException)) {
String heapDump = tryToDumpHeap();
LOG.error(
- "Execution of work for computation '{}' for sharding key '{}' failed
with out-of-memory. "
+ "Execution of work for fused stage '{}' for sharding key '{}' failed
with out-of-memory. "
+ "Work will not be retried locally. Heap dump {}.",
- computationId,
+ systemName,
work.getWorkItem().getShardingKey(),
heapDump,
parsedException);
@@ -177,8 +179,9 @@ public final class WorkFailureProcessor {
if (!failureTracker.trackFailure(computationId, work.getWorkItem(),
parsedException)) {
LOG.error(
- "Execution of work for computation '{}' on sharding key '{}' failed
with uncaught exception, "
+ "Execution of work for fused stage '{}' for computation '{}' on
sharding key '{}' failed with uncaught exception, "
+ "and Windmill indicated not to retry locally.",
+ systemName,
computationId,
work.getWorkItem().getShardingKey(),
parsedException);
@@ -186,10 +189,10 @@ public final class WorkFailureProcessor {
}
if
(elapsedTimeSinceStart.isLongerThan(MAX_LOCAL_PROCESSING_RETRY_DURATION)) {
LOG.error(
- "Execution of work for computation '{}' for sharding key '{}' failed
with uncaught exception, "
+ "Execution of work for fused stage '{}' for sharding key '{}' failed
with uncaught exception, "
+ "and it will not be retried locally because the elapsed time
since start {} "
+ "exceeds {}.",
- computationId,
+ systemName,
work.getWorkItem().getShardingKey(),
elapsedTimeSinceStart,
MAX_LOCAL_PROCESSING_RETRY_DURATION,
@@ -197,9 +200,9 @@ public final class WorkFailureProcessor {
return RetryEvaluation.DO_NOT_RETRY;
}
LOG.error(
- "Execution of work for computation '{}' on sharding key '{}' failed
with uncaught exception. "
+ "Execution of work for fused stage '{}' on sharding key '{}' failed
with uncaught exception. "
+ "Work will be retried locally.",
- computationId,
+ systemName,
work.getWorkItem().getShardingKey(),
parsedException);
return RetryEvaluation.RETRY_LOCALLY;
diff --git
a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/failures/WorkFailureProcessorTest.java
b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/failures/WorkFailureProcessorTest.java
index 89f3aa0c0d9..741cc35376f 100644
---
a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/failures/WorkFailureProcessorTest.java
+++
b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/failures/WorkFailureProcessorTest.java
@@ -50,6 +50,7 @@ import org.junit.runners.JUnit4;
public class WorkFailureProcessorTest {
private static final String DEFAULT_COMPUTATION_ID = "computationId";
+ private static final String DEFAULT_SYSTEM_NAME = "systemName";
private static WorkFailureProcessor createWorkFailureProcessor(
FailureTracker failureTracker, Supplier<Instant> clock) {
@@ -118,7 +119,11 @@ public class WorkFailureProcessorTest {
createWorkFailureProcessor(streamingEngineFailureReporter());
Set<Work> invalidWork = new HashSet<>();
workFailureProcessor.logAndProcessFailureBatch(
- DEFAULT_COMPUTATION_ID, List.of(work), new RuntimeException(),
invalidWork::add);
+ DEFAULT_COMPUTATION_ID,
+ DEFAULT_SYSTEM_NAME,
+ List.of(work),
+ new RuntimeException(),
+ invalidWork::add);
assertThat(executedWork).isEmpty();
assertThat(invalidWork).containsExactly(work.work());
@@ -135,7 +140,11 @@ public class WorkFailureProcessorTest {
OutOfMemoryError.class,
() ->
workFailureProcessor.logAndProcessFailureBatch(
- DEFAULT_COMPUTATION_ID, List.of(work), new OutOfMemoryError(),
invalidWork::add));
+ DEFAULT_COMPUTATION_ID,
+ DEFAULT_SYSTEM_NAME,
+ List.of(work),
+ new OutOfMemoryError(),
+ invalidWork::add));
assertThat(executedWork).isEmpty();
assertThat(invalidWork).isEmpty();
@@ -150,7 +159,11 @@ public class WorkFailureProcessorTest {
createWorkFailureProcessor(streamingApplianceFailureReporter(true));
Set<Work> invalidWork = new HashSet<>();
workFailureProcessor.logAndProcessFailureBatch(
- DEFAULT_COMPUTATION_ID, List.of(work), new RuntimeException(),
invalidWork::add);
+ DEFAULT_COMPUTATION_ID,
+ DEFAULT_SYSTEM_NAME,
+ List.of(work),
+ new RuntimeException(),
+ invalidWork::add);
assertThat(executedWork).isEmpty();
assertThat(invalidWork).containsExactly(work.work());
@@ -165,7 +178,11 @@ public class WorkFailureProcessorTest {
createWorkFailureProcessor(streamingEngineFailureReporter());
Set<Work> invalidWork = new HashSet<>();
workFailureProcessor.logAndProcessFailureBatch(
- DEFAULT_COMPUTATION_ID, List.of(veryOldWork), new RuntimeException(),
invalidWork::add);
+ DEFAULT_COMPUTATION_ID,
+ DEFAULT_SYSTEM_NAME,
+ List.of(veryOldWork),
+ new RuntimeException(),
+ invalidWork::add);
assertThat(executedWork).isEmpty();
assertThat(invalidWork).contains(veryOldWork.work());
@@ -180,7 +197,11 @@ public class WorkFailureProcessorTest {
createWorkFailureProcessor(streamingEngineFailureReporter());
Set<Work> invalidWork = new HashSet<>();
workFailureProcessor.logAndProcessFailureBatch(
- DEFAULT_COMPUTATION_ID, List.of(work), new RuntimeException(),
invalidWork::add);
+ DEFAULT_COMPUTATION_ID,
+ DEFAULT_SYSTEM_NAME,
+ List.of(work),
+ new RuntimeException(),
+ invalidWork::add);
runWork.await();
assertThat(invalidWork).isEmpty();
@@ -195,7 +216,11 @@ public class WorkFailureProcessorTest {
createWorkFailureProcessor(streamingApplianceFailureReporter(false));
Set<Work> invalidWork = new HashSet<>();
workFailureProcessor.logAndProcessFailureBatch(
- DEFAULT_COMPUTATION_ID, List.of(work), new RuntimeException(),
invalidWork::add);
+ DEFAULT_COMPUTATION_ID,
+ DEFAULT_SYSTEM_NAME,
+ List.of(work),
+ new RuntimeException(),
+ invalidWork::add);
runWork.await();
assertThat(invalidWork).isEmpty();
@@ -213,7 +238,11 @@ public class WorkFailureProcessorTest {
Set<Work> invalidWork = new HashSet<>();
workFailureProcessor.logAndProcessFailureBatch(
- DEFAULT_COMPUTATION_ID, List.of(work1, work2), new RuntimeException(),
invalidWork::add);
+ DEFAULT_COMPUTATION_ID,
+ DEFAULT_SYSTEM_NAME,
+ List.of(work1, work2),
+ new RuntimeException(),
+ invalidWork::add);
runWork1.await();
runWork2.await();
@@ -233,7 +262,11 @@ public class WorkFailureProcessorTest {
Set<Work> invalidWork = new HashSet<>();
workFailureProcessor.logAndProcessFailureBatch(
- DEFAULT_COMPUTATION_ID, List.of(work1, work2), new RuntimeException(),
invalidWork::add);
+ DEFAULT_COMPUTATION_ID,
+ DEFAULT_SYSTEM_NAME,
+ List.of(work1, work2),
+ new RuntimeException(),
+ invalidWork::add);
runWork1.await();
assertThat(executedWork2).isEmpty();
