Restore the behavior that failing to create a venv fails the pipeline. (

#26799)

CHANGES.md

@@ -257,6 +257,8 @@
 * If you activated a virtual environment in your custom container image, this environment might no longer be activated, since a new environment will be created (see the note about [BEAM-12792](https://issues.apache.org/jira/browse/BEAM-12792) above).
   To work around, install dependencies into the default (global) python environment. When using poetry you may need to use `poetry config virtualenvs.create false` before installing deps, see an example in: [#25085](https://github.com/apache/beam/issues/25085).
   If you were negatively impacted by this change and cannot find a workaround, feel free to chime in on [#16658](https://github.com/apache/beam/pull/16658).
+  To disable this behavior, you could upgrade to Beam 2.48.0 and set an environment variable
+  `ENV RUN_PYTHON_SDK_IN_DEFAULT_ENVIRONMENT=1` in your Dockerfile.
 
 ## Deprecations
 

sdks/python/container/boot.go

@@ -20,6 +20,7 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"flag"
 	"fmt"
 	"log"
@@ -161,14 +162,19 @@ func launchSDKProcess() error {
 	if os.Getenv("RUN_PYTHON_SDK_IN_DEFAULT_ENVIRONMENT") == "" {
 		venvDir, err := setupVenv(ctx, logger, "/opt/apache/beam-venv", *id)
 		if err != nil {
-			logger.Printf(ctx, "Using default environment, since creating a virtual environment for the SDK harness didn't succeed: %v", err)
-		} else {
-			cleanupFunc := func() {
-				os.RemoveAll(venvDir)
-				logger.Printf(ctx, "Cleaned up temporary venv for worker %v.", *id)
-			}
-			defer cleanupFunc()
+			return errors.New(
+				"failed to create a virtual environment. If running on Ubuntu systems, " +
+				"you might need to install `python3-venv` package. " +
+				"To run the SDK process in default environment instead, " +
+				"set the environment variable `RUN_PYTHON_SDK_IN_DEFAULT_ENVIRONMENT=1`. " +
+				"In custom Docker images, you can do that with an `ENV` statement. " +
+				fmt.Sprintf("Encountered error: %v", err))
+		}
+		cleanupFunc := func() {
+			os.RemoveAll(venvDir)
+			logger.Printf(ctx, "Cleaned up temporary venv for worker %v.", *id)
 		}
+		defer cleanupFunc()
 	}
 
 	dir := filepath.Join(*semiPersistDir, "staged")