Skip to content

Commit 5c0b302

Browse files
rwigglesrwiggles
committed
log the fused stage name instead of the computation name in more places
1 parent 9dab803 commit 5c0b302

8 files changed

Lines changed: 100 additions & 74 deletions

File tree

‎runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowWorkUnitClient.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -135,13 +135,13 @@ public Optional<WorkItem> getWorkItem() throws IOException {
135135

136136
final String stage;
137137
if (work.getMapTask() != null) {
138-
stage = work.getMapTask().getStageName();
138+
stage = work.getMapTask().getSystemName();
139139
logger.info("Starting MapTask stage {}", stage);
140140
} else if (work.getSeqMapTask() != null) {
141-
stage = work.getSeqMapTask().getStageName();
141+
stage = work.getSeqMapTask().getSystemName();
142142
logger.info("Starting SeqMapTask stage {}", stage);
143143
} else if (work.getSourceOperationTask() != null) {
144-
stage = work.getSourceOperationTask().getStageName();
144+
stage = work.getSourceOperationTask().getSystemName();
145145
logger.info("Starting SourceOperationTask stage {}", stage);
146146
} else {
147147
stage = null;

‎runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -267,6 +267,10 @@ public final long getBacklogBytes() {
267267
return backlogBytes;
268268
}
269269

270+
public String getSystemName() {
271+
return systemName;
272+
}
273+
270274
public long getMaxOutputKeyBytes() {
271275
return operationalLimits.getMaxOutputKeyBytes();
272276
}
@@ -584,7 +588,7 @@ public void invalidateCache() {
584588
} catch (IOException e) {
585589
Windmill.WorkItem workItem = getWorkItem();
586590
long shardingKey = workItem != null ? workItem.getShardingKey() : -1L;
587-
LOG.warn("Failed to close reader for {}-{}", computationId, shardingKey, e);
591+
LOG.warn("Failed to close reader for {}-{}", systemName, shardingKey, e);
588592
}
589593
}
590594
activeReader = null;

‎runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillSink.java‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -265,20 +265,26 @@ public long add(WindowedValue<T> data) throws IOException {
265265
}
266266
if (key.size() > context.getMaxOutputKeyBytes()) {
267267
if (context.throwExceptionsForLargeOutput()) {
268-
throw new OutputTooLargeException("Key too large: " + key.size());
268+
throw new OutputTooLargeException(
269+
String.format(
270+
"Key for system %s too large: %s", context.getSystemName(), key.size()));
269271
} else {
270272
LOG.error(
271-
"Trying to output too large key with size {}. Limit is {}. See https://cloud.google.com/dataflow/docs/guides/common-errors#key-commit-too-large-exception. Running with --experiments=throw_exceptions_on_large_output will instead throw an OutputTooLargeException which may be caught in user code.",
273+
"Trying to output too large key for system {} with size {}. Limit is {}. See https://cloud.google.com/dataflow/docs/guides/common-errors#key-commit-too-large-exception. Running with --experiments=throw_exceptions_on_large_output will instead throw an OutputTooLargeException which may be caught in user code.",
274+
context.getSystemName(),
272275
key.size(),
273276
context.getMaxOutputKeyBytes());
274277
}
275278
}
276279
if (value.size() > context.getMaxOutputValueBytes()) {
277280
if (context.throwExceptionsForLargeOutput()) {
278-
throw new OutputTooLargeException("Value too large: " + value.size());
281+
throw new OutputTooLargeException(
282+
String.format(
283+
"Value for system %s too large: %s", context.getSystemName(), value.size()));
279284
} else {
280285
LOG.error(
281-
"Trying to output too large value with size {}. Limit is {}. See https://cloud.google.com/dataflow/docs/guides/common-errors#key-commit-too-large-exception. Running with --experiments=throw_exceptions_on_large_output will instead throw an OutputTooLargeException which may be caught in user code.",
286+
"Trying to output too large value for system {} with size {}. Limit is {}. See https://cloud.google.com/dataflow/docs/guides/common-errors#key-commit-too-large-exception. Running with --experiments=throw_exceptions_on_large_output will instead throw an OutputTooLargeException which may be caught in user code.",
287+
context.getSystemName(),
282288
value.size(),
283289
context.getMaxOutputValueBytes());
284290
}

‎runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ComputationState.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,10 @@ public String getComputationId() {
6969
return computationId;
7070
}
7171

72+
public String getSystemName() {
73+
return mapTask.getSystemName();
74+
}
75+
7276
public MapTask getMapTask() {
7377
return mapTask;
7478
}

‎runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/Commit.java‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -66,9 +66,11 @@ public final String computationId() {
6666
return computationState().getComputationId();
6767
}
6868

69-
public @Nullable WorkItemCommitRequest singleKeyRequest() {
70-
return singleKeyRequest;
71-
};
69+
public final String systemName() {
70+
return computationState().getSystemName();
71+
}
72+
73+
public abstract WorkItemCommitRequest request();
7274

7375
public ComputationState computationState() {
7476
return computationState;

‎runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/commits/StreamingEngineWorkCommitter.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -112,7 +112,12 @@ public void commit(Commit commit) {
112112
// Do this check after adding to commitQueue, else commitQueue.put() can race with
113113
// drainCommitQueue() in stop() and leave commits orphaned in the queue.
114114
if (!this.isRunning.get()) {
115-
LOG.debug("Trying to queue commit on shutdown, failing commit={}", commit);
115+
LOG.debug(
116+
"Trying to queue commit on shutdown, failing commit=[systemName={}, shardingKey={},"
117+
+ " workId={} ].",
118+
commit.systemName(),
119+
commit.work().getShardedKey(),
120+
commit.work().id());
116121
drainCommitQueue();
117122
}
118123
}

‎runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java‎

Lines changed: 25 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -171,13 +171,13 @@ private static Windmill.WorkItemCommitRequest.Builder initializeOutputBuilder(
171171
}
172172

173173
/** Sets the stage name and workId of the Thread executing the {@link Work} for logging. */
174-
private static void setUpWorkLoggingContext(String workLatencyTrackingId, String computationId) {
174+
private static void setUpWorkLoggingContext(String workLatencyTrackingId, String systemName) {
175175
setLoggingContextWorkId(workLatencyTrackingId);
176-
setLoggingContextComputation(computationId);
176+
setLoggingContextSystemName(systemName);
177177
}
178178

179-
private static void setLoggingContextComputation(@Nullable String computationId) {
180-
DataflowWorkerLoggingMDC.setStageName(computationId);
179+
private static void setLoggingContextSystemName(@Nullable String systemName) {
180+
DataflowWorkerLoggingMDC.setStageName(systemName);
181181
}
182182

183183
private static void setLoggingContextWorkId(@Nullable String workLatencyTrackingId) {
@@ -227,15 +227,11 @@ public void queueAppliedFinalizeIds(ImmutableList<Long> appliedFinalizeIds) {
227227
private void processWork(
228228
ComputationState computationState, Work work, BoundedQueueExecutorWorkHandle handle) {
229229
Windmill.WorkItem workItem = work.getWorkItem();
230-
String computationId = computationState.getComputationId();
231-
LOG.debug("Starting processing for {}:\n{}", computationId, work);
232-
setLoggingContextComputation(computationId);
233-
KeyTransitionListener keyTransitionListener = createKeyTransitionListener();
234-
keyTransitionListener.onKeyTransition(null, work);
235-
236-
// Before any processing starts, call any pending OnCommit callbacks. Nothing that requires
237-
// cleanup should be done before this, since we might exit early here.
238-
commitFinalizer.finalizeCommits(workItem.getSourceState().getFinalizeIdsList());
230+
String systemName = computationState.getSystemName();
231+
work.setProcessingThreadName(Thread.currentThread().getName());
232+
work.setState(Work.State.PROCESSING);
233+
setUpWorkLoggingContext(work.getLatencyTrackingId(), systemName);
234+
LOG.debug("Starting processing for {}:\n{}", systemName, work);
239235

240236
if (workItem.getSourceState().getOnlyFinalize()) {
241237
handleOnlyFinalize(computationState, work, workItem);
@@ -264,15 +260,29 @@ private void processWork(
264260
recordProcessingStats(workBatch, workItemCommits, executeWorkResult.stateBytesRead());
265261
LOG.debug("Processing done for work batch size: {}", workBatch.size());
266262
} catch (Throwable t) {
267-
handleProcessWorkFailure(computationState, handle.getWorkBatch(), computationId, work, t);
263+
// OutOfMemoryError that are caught will be rethrown and trigger jvm termination.
264+
try {
265+
workFailureProcessor.logAndProcessFailure(
266+
systemName,
267+
ExecutableWork.create(work, (retry, h) -> processWork(computationState, retry, h)),
268+
t,
269+
invalidWork ->
270+
computationState.completeWorkAndScheduleNextWorkForKey(
271+
invalidWork.getShardedKey(), invalidWork.id()));
272+
} catch (OutOfMemoryError oom) {
273+
throw oom;
274+
} catch (Throwable t2) {
275+
LOG.warn("Failed to process work failure safely for work {}", work.id(), t2);
276+
throw ExceptionUtils.safeWrapThrowableAsException(t2);
277+
}
268278
} finally {
269279
List<Work> processedWorkBatch = workBatch != null ? workBatch : ImmutableList.of(work);
270280
// Update total processing time counters. Updating in finally clause ensures that
271281
// work items causing exceptions are also accounted in time spent.
272282
recordProcessingTime(stageInfo, processedWorkBatch, processingStartTimeNanos);
273283

274284
setLoggingContextWorkId(null);
275-
setLoggingContextComputation(null);
285+
setLoggingContextSystemName(null);
276286
sampler.resetForWorkId(work.getLatencyTrackingId());
277287
for (Work w : processedWorkBatch) {
278288
w.setProcessingThreadName("");

‎runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/failures/WorkFailureProcessor.java‎

Lines changed: 42 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -98,40 +98,25 @@ private static boolean isOutOfMemoryError(@Nullable Throwable t) {
9898
return false;
9999
}
100100

101-
public void logAndProcessFailureBatch(
102-
String computationId,
103-
List<ExecutableWork> executableWorks,
104-
Throwable t,
105-
Consumer<Work> onInvalidWork)
101+
/**
102+
* Processes failures caused by thrown exceptions that occur during execution of {@link Work}. May
103+
* attempt to retry execution of the {@link Work} or drop it if it is invalid.
104+
*/
105+
public void logAndProcessFailure(
106+
String systemName, ExecutableWork executableWork, Throwable t, Consumer<Work> onInvalidWork)
106107
throws Throwable {
107-
List<ExecutableWork> worksToRetryLocally = new java.util.ArrayList<>();
108-
109-
for (ExecutableWork executableWork : executableWorks) {
110-
switch (evaluateRetry(computationId, executableWork.work(), t)) {
111-
case DO_NOT_RETRY:
112-
// Consider the item invalid. It will eventually be retried by Windmill if it still needs
113-
// to be processed.
114-
onInvalidWork.accept(executableWork.work());
115-
break;
116-
case RETRY_LOCALLY:
117-
// Try again after some delay and at the end of the queue to avoid a tight loop.
118-
worksToRetryLocally.add(executableWork);
119-
break;
120-
case RETHROW_THROWABLE:
121-
throw t;
122-
}
123-
}
124-
125-
executeWithDelay(worksToRetryLocally);
126-
}
127-
128-
private void executeWithDelay(List<ExecutableWork> worksToRetryLocally) {
129-
if (!worksToRetryLocally.isEmpty()) {
130-
// Sleep ONCE for the entire batch delay to avoid sequential thread blocks
131-
Uninterruptibles.sleepUninterruptibly(retryLocallyDelayMs, TimeUnit.MILLISECONDS);
132-
for (ExecutableWork ew : worksToRetryLocally) {
133-
workUnitExecutor.forceExecute(ew, ew.work().getSerializedWorkItemSize());
134-
}
108+
switch (evaluateRetry(systemName, executableWork.work(), t)) {
109+
case DO_NOT_RETRY:
110+
// Consider the item invalid. It will eventually be retried by Windmill if it still needs to
111+
// be processed.
112+
onInvalidWork.accept(executableWork.work());
113+
break;
114+
case RETRY_LOCALLY:
115+
// Try again after some delay and at the end of the queue to avoid a tight loop.
116+
executeWithDelay(retryLocallyDelayMs, executableWork);
117+
break;
118+
case RETHROW_THROWABLE:
119+
throw t;
135120
}
136121
}
137122

@@ -148,12 +133,22 @@ private enum RetryEvaluation {
148133
RETHROW_THROWABLE,
149134
}
150135

151-
private RetryEvaluation evaluateRetry(String computationId, Work work, Throwable t) {
152-
if (work.isFailed()) {
136+
private RetryEvaluation evaluateRetry(String systemName, Work work, Throwable t) {
137+
@Nullable final Throwable cause = t.getCause();
138+
Throwable parsedException = (t instanceof UserCodeException && cause != null) ? cause : t;
139+
if (KeyTokenInvalidException.isKeyTokenInvalidException(parsedException)) {
140+
LOG.debug(
141+
"Execution of work for system '{}' on sharding key '{}' failed due to token expiration. "
142+
+ "Work will not be retried locally.",
143+
systemName,
144+
work.getWorkItem().getShardingKey());
145+
return RetryEvaluation.DO_NOT_RETRY;
146+
}
147+
if (WorkItemCancelledException.isWorkItemCancelledException(parsedException)) {
153148
LOG.debug(
154-
"Execution of work for computation '{}' on sharding key '{}' failed. "
155-
+ "Work is already marked as failed, not retrying locally.",
156-
computationId,
149+
"Execution of work for system '{}' on sharding key '{}' failed. "
150+
+ "Work will not be retried locally.",
151+
systemName,
157152
work.getWorkItem().getShardingKey());
158153
return RetryEvaluation.DO_NOT_RETRY;
159154
}
@@ -166,40 +161,40 @@ private RetryEvaluation evaluateRetry(String computationId, Work work, Throwable
166161
if (isOutOfMemoryError(parsedException)) {
167162
String heapDump = tryToDumpHeap();
168163
LOG.error(
169-
"Execution of work for computation '{}' for sharding key '{}' failed with out-of-memory. "
164+
"Execution of work for system '{}' for sharding key '{}' failed with out-of-memory. "
170165
+ "Work will not be retried locally. Heap dump {}.",
171-
computationId,
166+
systemName,
172167
work.getWorkItem().getShardingKey(),
173168
heapDump,
174169
parsedException);
175170
return RetryEvaluation.RETHROW_THROWABLE;
176171
}
177172

178-
if (!failureTracker.trackFailure(computationId, work.getWorkItem(), parsedException)) {
173+
if (!failureTracker.trackFailure(systemName, work.getWorkItem(), parsedException)) {
179174
LOG.error(
180-
"Execution of work for computation '{}' on sharding key '{}' failed with uncaught exception, "
175+
"Execution of work for system '{}' on sharding key '{}' failed with uncaught exception, "
181176
+ "and Windmill indicated not to retry locally.",
182-
computationId,
177+
systemName,
183178
work.getWorkItem().getShardingKey(),
184179
parsedException);
185180
return RetryEvaluation.DO_NOT_RETRY;
186181
}
187182
if (elapsedTimeSinceStart.isLongerThan(MAX_LOCAL_PROCESSING_RETRY_DURATION)) {
188183
LOG.error(
189-
"Execution of work for computation '{}' for sharding key '{}' failed with uncaught exception, "
184+
"Execution of work for system '{}' for sharding key '{}' failed with uncaught exception, "
190185
+ "and it will not be retried locally because the elapsed time since start {} "
191186
+ "exceeds {}.",
192-
computationId,
187+
systemName,
193188
work.getWorkItem().getShardingKey(),
194189
elapsedTimeSinceStart,
195190
MAX_LOCAL_PROCESSING_RETRY_DURATION,
196191
parsedException);
197192
return RetryEvaluation.DO_NOT_RETRY;
198193
}
199194
LOG.error(
200-
"Execution of work for computation '{}' on sharding key '{}' failed with uncaught exception. "
195+
"Execution of work for system '{}' on sharding key '{}' failed with uncaught exception. "
201196
+ "Work will be retried locally.",
202-
computationId,
197+
systemName,
203198
work.getWorkItem().getShardingKey(),
204199
parsedException);
205200
return RetryEvaluation.RETRY_LOCALLY;

0 commit comments

Comments
 (0)