Skip to content

Log the System name in more places instead of the computationId #39665

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rwiggles wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Log the System name in more places instead of the computationId #39665

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2d2915f
Switch StreamingWorkScheduler and WorkFailureProcessor logging to sys…
rwiggles Aug 6, 2026
df20eba
Switch StreamingWorkScheduler and WorkFailureProcessor logging to sys…
rwiggles Aug 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions .../apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -227,8 +227,9 @@ private void processWork(
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);

Expand Down Expand Up @@ -259,7 +260,8 @@ private void processWork(
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
Expand Down Expand Up @@ -453,6 +455,7 @@ private void handleProcessWorkFailure(
ComputationState computationState,
List<Work> failedBatch,
String computationId,
String systemName,
Work primaryWork,
Throwable t) {
try {
Expand All @@ -464,6 +467,7 @@ private void handleProcessWorkFailure(

workFailureProcessor.logAndProcessFailureBatch(
computationId,
systemName,
executableWorks,
t,
invalidWork ->
Expand Down
38 changes: 25 additions & 13 deletions .../beam/runners/dataflow/worker/windmill/work/processing/failures/WorkFailureProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,14 +100,15 @@ private static boolean isOutOfMemoryError(@Nullable Throwable t) {

public void logAndProcessFailureBatch(
String computationId,
String systemName,
List<ExecutableWork> executableWorks,
Throwable t,
Consumer<Work> onInvalidWork)
throws Throwable {
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.
Expand Down Expand Up @@ -148,12 +149,15 @@ private enum RetryEvaluation {
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. "
"Execution of work for fused stage '{}' on sharding key '{}' failed. "
+ "Work is already marked as failed, not retrying locally.",
computationId,
systemName,
systemName,
work.getWorkItem().getShardingKey());
return RetryEvaluation.DO_NOT_RETRY;
}
Expand All @@ -166,40 +170,48 @@ private RetryEvaluation evaluateRetry(String computationId, Work work, Throwable
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. "
"Execution of work for fused stage '{}' for sharding key '{}' failed with out-of-memory. "
+ "Work will not be retried locally. Heap dump {}.",
computationId,
systemName,
systemName,
work.getWorkItem().getShardingKey(),
heapDump,
parsedException);
return RetryEvaluation.RETHROW_THROWABLE;
}

if (!failureTracker.trackFailure(computationId, work.getWorkItem(), parsedException)) {
if (!failureTracker.trackFailure(systemName, work.getWorkItem(), parsedException)) {
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, "
"Execution of work for fused stage '{}' on sharding key '{}' failed with uncaught exception, "
+ "and Windmill indicated not to retry locally.",
computationId,
systemName,
systemName,
work.getWorkItem().getShardingKey(),
parsedException);
return RetryEvaluation.DO_NOT_RETRY;
}
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, "
"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,
systemName,
work.getWorkItem().getShardingKey(),
elapsedTimeSinceStart,
MAX_LOCAL_PROCESSING_RETRY_DURATION,
parsedException);
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. "
"Execution of work for fused stage '{}' on sharding key '{}' failed with uncaught exception. "
+ "Work will be retried locally.",
computationId,
systemName,
systemName,
work.getWorkItem().getShardingKey(),
parsedException);
return RetryEvaluation.RETRY_LOCALLY;
Expand Down
49 changes: 41 additions & 8 deletions ...m/runners/dataflow/worker/windmill/work/processing/failures/WorkFailureProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@
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) {
Expand Down Expand Up @@ -118,7 +119,11 @@ public void logAndProcessFailureBatch_doesNotRetryFailedWork() throws Throwable
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());
Expand All @@ -135,7 +140,11 @@ public void logAndProcessFailureBatch_doesNotRetryOOM() {
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();
Expand All @@ -150,7 +159,11 @@ public void logAndProcessFailureBatch_doesNotRetryWhenFailureReporterMarksAsNonR
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());
Expand All @@ -165,7 +178,11 @@ public void logAndProcessFailureBatch_doesNotRetryAfterLocalRetryTimeout() throw
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());
Expand All @@ -180,7 +197,11 @@ public void logAndProcessFailureBatch_retriesOnUncaughtUnhandledException_stream
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();
Expand All @@ -195,7 +216,11 @@ public void logAndProcessFailureBatch_retriesOnUncaughtUnhandledException_stream
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();
Expand All @@ -213,7 +238,11 @@ public void logAndProcessFailureBatch_retryAll() throws Throwable {
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();
Expand All @@ -233,7 +262,11 @@ public void logAndProcessFailureBatch_mixRetryAndAbort() throws Throwable {
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();
Expand Down