arvindram03 commented on code in PR #29882:
URL: https://github.com/apache/beam/pull/29882#discussion_r1446541244
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOperationContext.java:
##########
@@ -312,6 +329,45 @@ public void reportLull(Thread trackedThread, long millis) {
}
}
+ @Override
+ public void reportBundleLull(Thread trackedThread, String
customLogMessage, long millis) {
+ // If we're not logging warnings, nothing to report.
+ if (!LOG.isWarnEnabled()) {
+ return;
+ }
+
+ Duration lullDuration = Duration.millis(millis);
+
+ // Since the lull reporting executes in the sampler thread, it won't
automatically inherit the
+ // context of the current step. To ensure things are logged correctly,
we get the currently
+ // registered DataflowWorkerLoggingHandler and log directly in the
desired context.
+ LogRecord logRecord =
+ new LogRecord(Level.WARNING, getBundleLullMessage(trackedThread,
lullDuration));
+ logRecord.setLoggerName(DataflowOperationContext.LOG.getName());
+ LogRecord customLogRecord = new LogRecord(Level.WARNING,
customLogMessage);
+ customLogRecord.setLoggerName(DataflowOperationContext.LOG.getName());
+
+ // Publish directly in the context of this specific ExecutionState.
+ DataflowWorkerLoggingHandler dataflowLoggingHandler =
+ DataflowWorkerLoggingInitializer.getLoggingHandler();
+ dataflowLoggingHandler.publish(this, logRecord);
+ dataflowLoggingHandler.publish(this, customLogRecord);
+
+ if (shouldLogFullThreadDumpForBundle(lullDuration)) {
+ Map<Thread, StackTraceElement[]> threadSet =
Thread.getAllStackTraces();
+ for (Map.Entry<Thread, StackTraceElement[]> entry :
threadSet.entrySet()) {
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
