update description and logs for timeout (#36003)

* update description and logs for timeout

* minor fix

* go format fix

* fix

* Update changes.md

* Update the flag description in release note

* remove whitespace

* Update CHANGES.md

slightly updated the description

* Update CHANGES.md

---------

Co-authored-by: tvalentyn <tvalentyn@users.noreply.github.com>

Author: LEEKYE

CHANGES.md

@@ -81,6 +81,9 @@
   Beam now supports data enrichment capabilities using SQL databases, with built-in support for:
   - Managed PostgreSQL, MySQL, and Microsoft SQL Server instances on CloudSQL
   - Unmanaged SQL database instances not hosted on CloudSQL (e.g., self-hosted or on-premises databases)
+* Added a pipeline option to specify the processing timeout for a single element by any PTransform (Java/Python/Go) ([#35174](https://github.com/apache/beam/issues/35174)).
+  - When specified, the SDK harness automatically restarts if an element takes too long to process. Beam runner may then retry processing of the same work item.
+  - Use the `--element_processing_timeout_minutes` option to reduce the chance of having stalled pipelines due to unexpected cases of slow processing, where slowness might not happen again if processing of the same element is retried.
 
 ## Breaking Changes
 

sdks/go/pkg/beam/core/metrics/sampler.go

@@ -63,11 +63,11 @@ func (s *StateSampler) Sample(ctx context.Context, t time.Duration) error {
 		}
 
 		if s.millisSinceLastTransition > s.nextLogTime {
-			log.Infof(ctx, "Operation ongoing in transform %v for at least %v ms without outputting or completing in state %v", ps.pid, s.millisSinceLastTransition, getState(ps.state))
+			log.Infof(ctx, "Operation ongoing in transform %v for at least %v without outputting or completing in state %v", ps.pid, s.millisSinceLastTransition, getState(ps.state))
 			s.nextLogTime += s.logInterval
 		}
 		if s.restartLullTimeout > 0 && s.millisSinceLastTransition > s.restartLullTimeout {
-			return errors.Errorf("Operation ongoing in transform %v for at least %v ms without outputting or completing in state %v, the SDK harness will be terminated and restarted", ps.pid, s.millisSinceLastTransition, getState(ps.state))
+			return errors.Errorf("Processing of an element in transform %v has exceeded the specified timeout of %v without outputting or completing in state %v, SDK harness will be terminated", ps.pid, s.restartLullTimeout, getState(ps.state))
 		}
 	}
 	return nil

sdks/go/pkg/beam/core/metrics/sampler_test.go

@@ -176,8 +176,8 @@ func TestSamplerWithRestartLullTimeout(t *testing.T) {
 		t.Errorf("s.sample(bctx, interval) = %v, want %v", got, want)
 	}
 	err := s.Sample(bctx, interval)
-	if err == nil || !strings.Contains(err.Error(), "the SDK harness will be terminated and restarted") {
-		t.Errorf("s.sample(bctx, interval) = %v, want %v", err, "the SDK harness will be terminated and restarted")
+	if err == nil || !strings.Contains(err.Error(), "SDK harness will be terminated") {
+		t.Errorf("s.sample(bctx, interval) = %v, want %v", err, "SDK harness will be terminated")
 	}
 }
 

sdks/go/pkg/beam/options/jobopts/options.go

@@ -102,7 +102,9 @@ var (
 	ResourceHints stringSlice
 
 	// ElementProcessingTimeout flag to set the timeout for processing an element in a PTransform operation. If set to -1, there is no timeout.
-	ElementProcessingTimeout = flag.Duration("element_processing_timeout", -1, "The timeout for processing an element in a PTransform operation. If set to -1, there is no timeout.")
+	ElementProcessingTimeout = flag.Duration("element_processing_timeout", -1,
+		"The time limit (in minutes) for any PTransform to finish processing a single element. If exceeded, "+
+			"the SDK worker process self-terminates and processing may be restarted by a runner. There is no time limit if the value is set to -1.")
 )
 
 type missingFlagError error

sdks/java/core/src/main/java/org/apache/beam/sdk/options/SdkHarnessOptions.java

@@ -433,8 +433,9 @@ public Duration create(PipelineOptions options) {
    * signaling the runner harness to restart the SDK worker.
    */
   @Description(
-      "The time limit (minute) that an SDK worker allows for a PTransform operation "
-          + "before signaling the runner harness to restart the SDK worker. There is no time limit if the value is set to 0.")
+      "The time limit (in minutes) for any PTransform to finish processing a single element."
+          + " If exceeded, the SDK worker process self-terminates and processing may be restarted by a runner."
+          + " There is no time limit if the value is set to 0.")
   @NonNegative
   int getElementProcessingTimeoutMinutes();
 

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ExecutionStateSampler.java

@@ -400,25 +400,26 @@ private Optional<String> takeSample(long currentTimeMillis, long millisSinceLast
           if (thread == null) {
             timeoutMessage =
                 String.format(
-                    "Operation ongoing in bundle %s for at least %s without outputting "
-                        + "or completing (stack trace unable to be generated). The SDK worker will restart.",
+                    "Processing of an element in bundle %s has exceeded the specified timeout of %s "
+                        + "(stack trace unable to be generated). The SDK worker will be terminated.",
                     processBundleId.get(),
                     DURATION_FORMATTER.print(
                         Duration.millis(userSpecifiedLullTimeMsForRestart).toPeriod()));
           } else if (currentExecutionState == null) {
             timeoutMessage =
                 String.format(
-                    "Operation ongoing in bundle %s for at least %s without outputting "
-                        + "or completing:%n  at %s. The SDK worker will restart.",
+                    "Processing of an element in bundle %s has exceeded the specified timeout of %s "
+                        + "without outputting or completing:%n  at %s. The SDK worker will be terminated.",
                     processBundleId.get(),
                     DURATION_FORMATTER.print(
                         Duration.millis(userSpecifiedLullTimeMsForRestart).toPeriod()),
                     Joiner.on("\n  at ").join(thread.getStackTrace()));
           } else {
             timeoutMessage =
                 String.format(
-                    "Operation ongoing in bundle %s for PTransform{id=%s, name=%s, state=%s} "
-                        + "for at least %s without outputting or completing:%n  at %s. The SDK worker will restart.",
+                    "Processing of an element in bundle %s for PTransform{id=%s, name=%s, state=%s} "
+                        + "has exceeded the specified timeout of %s without outputting or completing:%n  at %s. "
+                        + "The SDK worker will be terminated.",
                     processBundleId.get(),
                     currentExecutionState.ptransformId,
                     currentExecutionState.ptransformUniqueName,

sdks/python/apache_beam/options/pipeline_options.py

@@ -1461,9 +1461,10 @@ def _add_argparse_args(cls, parser):
         type=int,
         default=None,
         help=(
-            'The time limit (in minutes) that an SDK worker allows for a'
-            ' PTransform operation to process one element before signaling'
-            ' the runner harness to restart the SDK worker.'))
+            'The time limit (in minutes) for any PTransform to finish '
+            'processing a single element. If exceeded, the SDK worker '
+            'process self-terminates and processing may be restarted '
+            'by a runner.'))
 
   def validate(self, validator):
     errors = []

sdks/python/apache_beam/runners/worker/worker_status.py

@@ -185,7 +185,8 @@ def __init__(
         self._status_channel)
     self._responses = queue.Queue()
     self.log_lull_timeout_ns = log_lull_timeout_ns
-    if element_processing_timeout_minutes:
+    if (element_processing_timeout_minutes and
+        element_processing_timeout_minutes > 0):
       self._element_processing_timeout_ns = (
           element_processing_timeout_minutes * 60 * 1e9)
     else:
@@ -285,12 +286,13 @@ def _log_lull_sampler_info(self, sampler_info, instruction):
     if timeout_exceeded:
       _LOGGER.error(
           (
-              'Operation ongoing in bundle %s%s for at least %.2f seconds'
-              ' without outputting or completing.\n'
+              'Processing of an element in bundle %s%s has exceeded the '
+              'specified timeout of %.2f minutes. SDK harness will be '
+              'terminated.\n'
               'Current Traceback:\n%s'),
           instruction,
           step_name_log,
-          lull_seconds,
+          self._element_processing_timeout_ns / 1e9 / 60,
           stack_trace,
       )
       from apache_beam.runners.worker.sdk_worker_main import terminate_sdk_harness