[Python] Support large pipeline options via file (Fixes #37370)

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

@@ -93,8 +93,25 @@ def create_harness(environment, dry_run=False):
   else:
     fn_log_handler = None
 
-  pipeline_options_dict = _load_pipeline_options(
-      environment.get('PIPELINE_OPTIONS'))
+  options_json = environment.get('PIPELINE_OPTIONS')
+
+  #We check if options are stored in the file.
+  if 'PIPELINE_OPTIONS_FILE' in environment:
+    options_file = environment['PIPELINE_OPTIONS_FILE']
+    try:
+      with open(options_file, 'r') as f:
+        options_json = f.read()
+        _LOGGER.info('Load pipeline options from file: %s', options_file)
+    except:
+      _LOGGER.error(
+          'Failed to load pipeline options from file: %s', options_file)
+      raise
+
+  pipeline_options_dict = _load_pipeline_options(options_json)
+
+  #pipeline_options_dict = _load_pipeline_options(
+  #    environment.get('PIPELINE_OPTIONS'))
+
   default_log_level = _get_log_level_from_options_dict(pipeline_options_dict)
   logging.getLogger().setLevel(default_log_level)
   _set_log_level_overrides(pipeline_options_dict)

sdks/python/container/boot.go

@@ -220,7 +220,18 @@ func launchSDKProcess() error {
 
 	// (3) Invoke python
 
-	os.Setenv("PIPELINE_OPTIONS", options)
+	//Commented out -->
+	//os.Setenv("PIPELINE_OPTIONS", options)
+
+	//To prevent crashes if options > ARG_MAX -->
+	// --> Write the large JSON content into a file on disk
+	// --> environment only receives the short filename
+	if optionsFile, err := MakePipelineOptionsFileAndEnvVar(options); err != nil {
+		logger.Fatalf(ctx, "Failed to create a pipeline options file: %v", err)
+	} else {
+		os.Setenv("PIPELINE_OPTIONS_FILE", optionsFile)
+	}
+	os.Setenv("SEMI_PERSISTENT_DIRECTORY", *semiPersistDir)
 	os.Setenv("SEMI_PERSISTENT_DIRECTORY", *semiPersistDir)
 	os.Setenv("LOGGING_API_SERVICE_DESCRIPTOR", (&pipepb.ApiServiceDescriptor{Url: *loggingEndpoint}).String())
 	os.Setenv("CONTROL_API_SERVICE_DESCRIPTOR", (&pipepb.ApiServiceDescriptor{Url: *controlEndpoint}).String())
@@ -503,3 +514,21 @@ func logSubmissionEnvDependencies(ctx context.Context, bufLogger *tools.Buffered
 	bufLogger.Printf(ctx, "%s", string(content))
 	return nil
 }
+
+// MakePipelineOptionsFileAndEnvVar writes the pipeline options to a file.
+// Assumes the options string is JSON formatted.
+//
+// Stores the file name in question in PIPELINE_OPTIONS_FILE for access by the SDK.
+func MakePipelineOptionsFileAndEnvVar(options string) (string, error) {
+	fn := "pipeline_options.json"
+	f, err := os.Create(fn)
+	if err != nil {
+		return "", fmt.Errorf("unable to create %v: %w", fn, err)
+	}
+	defer f.Close()
+	if _, err := f.WriteString(options); err != nil {
+		return "", fmt.Errorf("error writing %v: %w", f.Name(), err)
+	}
+	os.Setenv("PIPELINE_OPTIONS_FILE", f.Name())
+	return fn, nil
+}