From de0231ca04f051c58f76084e63e658548a18840f Mon Sep 17 00:00:00 2001 From: Prasad Lohakpure Date: Thu, 30 Jul 2026 15:49:51 +0530 Subject: [PATCH 1/3] FEAT: Support for custom entrypoint in sparkeks Enable the sparkeks plugin to run custom-class (Scala/Java) Spark applications (e.g. com.pattern.chipmunk.Writer) alongside the existing Python SQL wrapper. - entrypoint.go: entrypoint strategy pattern selected by wrapper_uri extension (.jar -> JAR/--class, .py/default -> SQL wrapper). JAR mode sets Spec.Type (Scala/Java via parameters.application_type, default Scala) and MainClass from parameters.entry_point. - Managed positional args [appName, queryURI, user, (resultURI)] shared by both entrypoints; job context `arguments`, when set, is passed to the app verbatim so callers can match an app's exact CLI contract. - Precompute s3a query/result URIs once on the execution context. - Write submission-failure / unknown-state reason to job stderr. - Default Spec.Type=Python guards templateless jobs. - plugins/sparkeks/README.md: document JAR entrypoint support, extension dispatch, entry_point/application_type, and the argument contract. --- .../pkg/object/command/sparkeks/entrypoint.go | 131 ++++++++++++++++++ .../pkg/object/command/sparkeks/sparkeks.go | 40 ++++-- plugins/sparkeks/README.md | 85 ++++++++++++ 3 files changed, 245 insertions(+), 11 deletions(-) create mode 100644 internal/pkg/object/command/sparkeks/entrypoint.go diff --git a/internal/pkg/object/command/sparkeks/entrypoint.go b/internal/pkg/object/command/sparkeks/entrypoint.go new file mode 100644 index 00000000..7097f7b1 --- /dev/null +++ b/internal/pkg/object/command/sparkeks/entrypoint.go @@ -0,0 +1,131 @@ +package sparkeks + +import ( + "path" + "strings" + + "github.com/kubeflow/spark-operator/v2/api/v1beta2" +) + +// --- JAR application type (JVM language) resolution --- +// Add an entry to support another JVM application type. +//e.g. Combination: applicationType (Java) -> EntrypointStrategy (JAR) + (Class name) + +var jarApplicationTypes = map[string]v1beta2.SparkApplicationType{ + "scala": v1beta2.SparkApplicationTypeScala, + "java": v1beta2.SparkApplicationTypeJava, +} + +const defaultJarApplicationType = v1beta2.SparkApplicationTypeScala + +// resolveJarApplicationType maps an optional configured application type (e.g. "Java", "Scala", +// case-insensitive) to its SparkApplication type, falling back to the default (Scala) when +// unset or unrecognized. +func resolveJarApplicationType(applicationType string) v1beta2.SparkApplicationType { + if t, ok := jarApplicationTypes[strings.ToLower(strings.TrimSpace(applicationType))]; ok { + return t + } + return defaultJarApplicationType +} + +type entrypointStrategy interface { + apply(spec *v1beta2.SparkApplicationSpec) +} + +func buildArguments(override []string, appName, queryURI, user, resultURI string, returnResult bool) []string { + if len(override) > 0 { + return override + } + args := []string{appName, queryURI, user} + if returnResult { + args = append(args, resultURI) + } + return args +} + +type jarEntrypointStrategy struct { + appType v1beta2.SparkApplicationType + mainClass string + appName string + queryURI string + user string + resultURI string + returnResult bool + arguments []string +} + +func (s jarEntrypointStrategy) apply(spec *v1beta2.SparkApplicationSpec) { + mainClass := s.mainClass + spec.Type = s.appType + spec.MainClass = &mainClass + + spec.Arguments = buildArguments(s.arguments, s.appName, s.queryURI, s.user, s.resultURI, s.returnResult) +} + +type sqlWrapperEntrypointStrategy struct { + appName string + queryURI string + user string + resultURI string + returnResult bool + arguments []string +} + +func (s sqlWrapperEntrypointStrategy) apply(spec *v1beta2.SparkApplicationSpec) { + spec.Arguments = buildArguments(s.arguments, s.appName, s.queryURI, s.user, s.resultURI, s.returnResult) +} + +// --- Extension-based strategy selection --- + +// entrypointFactory builds the entrypoint strategy for a job from its execution context. +type entrypointFactory func(execCtx *executionContext) entrypointStrategy + +var entrypointStrategiesByExt = map[string]entrypointFactory{ + ".jar": newJarEntrypointStrategy, + ".py": newSQLWrapperEntrypointStrategy, +} + +var defaultEntrypointFactory entrypointFactory = newSQLWrapperEntrypointStrategy + +func newEntrypointStrategy(execCtx *executionContext) entrypointStrategy { + ext := strings.ToLower(path.Ext(execCtx.commandContext.WrapperURI)) + factory, ok := entrypointStrategiesByExt[ext] + if !ok { + factory = defaultEntrypointFactory + } + return factory(execCtx) +} + +func newJarEntrypointStrategy(execCtx *executionContext) entrypointStrategy { + jobContext := execCtx.jobContext + + var mainClass, applicationType string + if jobContext.Parameters != nil { + mainClass = jobContext.Parameters.EntryPoint + applicationType = jobContext.Parameters.ApplicationType + } + + return jarEntrypointStrategy{ + appType: resolveJarApplicationType(applicationType), + mainClass: mainClass, + appName: execCtx.appName, + queryURI: execCtx.s3aQueryURI, + user: execCtx.job.User, + resultURI: execCtx.s3aResultURI, + returnResult: jobContext.ReturnResult, + arguments: jobContext.Arguments, + } +} + +func newSQLWrapperEntrypointStrategy(execCtx *executionContext) entrypointStrategy { + jobContext := execCtx.jobContext + + return sqlWrapperEntrypointStrategy{ + appName: execCtx.appName, + queryURI: execCtx.s3aQueryURI, + user: execCtx.job.User, + resultURI: execCtx.s3aResultURI, + returnResult: jobContext.ReturnResult, + arguments: jobContext.Arguments, + } +} diff --git a/internal/pkg/object/command/sparkeks/sparkeks.go b/internal/pkg/object/command/sparkeks/sparkeks.go index 909561ef..a9f41c80 100644 --- a/internal/pkg/object/command/sparkeks/sparkeks.go +++ b/internal/pkg/object/command/sparkeks/sparkeks.go @@ -52,6 +52,11 @@ const ( defaultSparkAppAPIVersion = "sparkoperator.k8s.io/v1beta2" defaultSparkAppKind = "SparkApplication" + // defaultApplicationType is applied when neither the loaded template nor the JAR branch + // set Spec.Type, so the CRD's required enum field is never submitted empty (see Gap 4). + // JAR application types (jarApplicationTypes / resolveJarApplicationType) live in entrypoint.go. + defaultApplicationType = v1beta2.SparkApplicationTypePython + queriesPath = "queries" resultsPath = "results" logsPath = "logs" @@ -104,10 +109,13 @@ type commandContext struct { type jobParameters struct { Properties map[string]string `yaml:"properties,omitempty" json:"properties,omitempty"` + EntryPoint string `yaml:"entry_point,omitempty" json:"entry_point,omitempty"` + ApplicationType string `yaml:"application_type,omitempty" json:"application_type,omitempty"` } type jobContext struct { Query string `yaml:"query,omitempty" json:"query,omitempty"` + Arguments []string `yaml:"arguments,omitempty" json:"arguments,omitempty"` Parameters *jobParameters `yaml:"parameters,omitempty" json:"parameters,omitempty"` ReturnResult bool `yaml:"return_result,omitempty" json:"return_result,omitempty"` } @@ -139,6 +147,8 @@ type executionContext struct { appName string queryURI string resultURI string + s3aQueryURI string + s3aResultURI string logURI string sparkApp *v1beta2.SparkApplication submittedApp *v1beta2.SparkApplication @@ -370,6 +380,8 @@ func buildExecutionContextAndURI(ctx context.Context, r *plugin.Runtime, j *job. execCtx.appName = fmt.Sprintf("%s-%s", applicationPrefix, j.ID) execCtx.queryURI = fmt.Sprintf("%s/%s/%s/%s", s.JobsURI, j.ID, queriesPath, queryFileName) execCtx.resultURI = fmt.Sprintf("%s/%s/%s", s.JobsURI, j.ID, resultsPath) + execCtx.s3aQueryURI = updateS3ToS3aURI(execCtx.queryURI) + execCtx.s3aResultURI = updateS3ToS3aURI(execCtx.resultURI) execCtx.logURI = fmt.Sprintf("%s/%s/%s", s.JobsURI, j.ID, logsPath) // Upload query to S3 @@ -472,6 +484,7 @@ func updateS3ToS3aURI(uri string) string { return strings.ReplaceAll(uri, s3Prefix, s3aPrefix) } + // getS3FileURI finds a file in an S3 directory that matches the given extension. func getS3FileURI(ctx context.Context, awsConfig aws.Config, directoryURI, matchingExtension string) (string, error) { s3Parts := rxS3.FindAllStringSubmatch(directoryURI, -1) @@ -727,18 +740,14 @@ func applySparkOperatorConfig(execCtx *executionContext) { sparkApp.ObjectMeta.Name = execCtx.appName sparkApp.ObjectMeta.Namespace = execCtx.commandContext.KubeNamespace - // Set main application file and arguments + // Set the main application file (common to both entrypoint styles), then delegate the + // entrypoint-specific spec (Type / MainClass / Arguments) to a strategy: JAR/--class vs the + // default SQL wrapper (see entrypointStrategy above). if execCtx.commandContext.WrapperURI != "" { s3aWrapperURI := updateS3ToS3aURI(execCtx.commandContext.WrapperURI) - s3aQueryURI := updateS3ToS3aURI(execCtx.queryURI) - s3aResultURI := updateS3ToS3aURI(execCtx.resultURI) - mainAppFile := s3aWrapperURI - if jobContext.ReturnResult { - sparkApp.Spec.Arguments = []string{execCtx.appName, s3aQueryURI, execCtx.job.User, s3aResultURI} - } else { - sparkApp.Spec.Arguments = []string{execCtx.appName, s3aQueryURI, execCtx.job.User} - } - sparkApp.Spec.MainApplicationFile = &mainAppFile + sparkApp.Spec.MainApplicationFile = &s3aWrapperURI + // Type / MainClass / Arguments are set by the entrypoint strategy. + newEntrypointStrategy(execCtx).apply(&sparkApp.Spec) } if sparkApp.Spec.SparkConf == nil { @@ -844,6 +853,10 @@ func applySparkOperatorConfig(execCtx *executionContext) { for k, v := range jobContext.Parameters.Properties { sparkApp.Spec.SparkConf[k] = v } + + if sparkApp.Spec.Type == "" { + sparkApp.Spec.Type = defaultApplicationType + } } // generateSparkApp generates the Spark application from the template. @@ -967,7 +980,12 @@ func (e *executionContext) monitorJobAndCollectLogs(ctx context.Context) error { if state == v1beta2.ApplicationStateUnknown { msg = sparkAppUnknownStateMsg } - return fmt.Errorf("%s", msg) + errorMessage := sparkApp.Status.AppState.ErrorMessage + if errorMessage == "" { + errorMessage = unknownErrorMsg + } + e.runtime.Stderr.WriteString(fmt.Sprintf("%s: %s\n", msg, errorMessage)) + return fmt.Errorf("%s: %s", msg, errorMessage) } } } diff --git a/plugins/sparkeks/README.md b/plugins/sparkeks/README.md index 0c923947..7a66956b 100644 --- a/plugins/sparkeks/README.md +++ b/plugins/sparkeks/README.md @@ -10,6 +10,7 @@ The Spark EKS plugin enables submitting Spark jobs to an AWS EKS cluster using t - Supports custom SparkApplication YAML templates - IAM role assumption for cross-account EKS access - Configurable Spark job resources and properties +- **JAR / `--class` entrypoints** — run a JVM (Scala/Java) application by main class, alongside the default Python SQL wrapper ## Configuration @@ -59,6 +60,90 @@ The Spark EKS plugin enables submitting Spark jobs to an AWS EKS cluster using t } ``` +## Entrypoints: SQL wrapper vs JAR + +The plugin picks an entrypoint from the **file extension of the command's `wrapper_uri`**. The wrapper +file itself is always submitted as `MainApplicationFile`; only `Type`, `MainClass`, and `Arguments` +differ between the two. + +| `wrapper_uri` ends in | Entrypoint | `Spec.Type` | `Spec.MainClass` | +|---|---|---|---| +| `.py` | Python SQL wrapper (default, pre-existing behavior) | `Python` | not set | +| `.jar` | JVM JAR, run by main class | `Scala` or `Java` | `entry_point` | +| anything else | falls back to the SQL wrapper | `Python` | not set | + +### JAR configuration + +Set on the **job context** under `parameters`: + +| Field | Required | Description | +|---|---|---| +| `entry_point` | yes (for JAR) | Fully-qualified main class, e.g. `com.org.customapplication.main`. Becomes `--class`. | +| `application_type` | no | `Scala` or `Java`, case-insensitive. Defaults to `Scala` when empty or unrecognized. | + +```json +{ + "query": "SELECT key, value FROM my_table", + "wrapper_uri": "s3://mybucket/application-assembly.jar", + "parameters": { + "entry_point": "com.org.customapplication.main", + "application_type": "Scala", + "properties": { + //sparkeks properties + } + }, + "return_result": false +} +``` + +### Arguments + +If the job context sets `arguments`, that list is passed to the app **verbatim** and the managed +list is not used — letting callers match an app's exact CLI contract (e.g. an app that expects +`[app_name, user, query, s3_path]`). Jobs that set no `arguments` keep the managed behavior, so +existing contracts are unchanged. + + +#### Example 1 — context provided + +No `arguments`; the SQL travels as an uploaded `queryURI`: + +```json +{ + "context": { + "query": "SELECT * FROM my_table LIMIT 10;", + "return_result": true + } +} +``` + +The app receives `[appName, query, user, resultURI]`. + +#### Example 2 — verbatim override + +`arguments` supplies the app's exact CLI shape (here `[app_name, user, query, s3_path]`); the +managed list is ignored: + +```json +{ + "context": { + "arguments": [ + "my-app", + "some_user", + "SELECT key, value FROM my_table", + "s3://mybucket/output/v1" + ], + "parameters": { + "entry_point": "com.org.customapplication.main", + "application_type": "Scala" + } + } +} +``` + +Here the app receives exactly those four strings. The SQL is passed inline, so mind arg-length +limits for very large queries. + ## Usage Submit a job using the API: From 8114fc94a5b361fa60899b43ff16fddd8615c54b Mon Sep 17 00:00:00 2001 From: Prasad Lohakpure Date: Thu, 30 Jul 2026 20:51:23 +0530 Subject: [PATCH 2/3] Reduced comments --- internal/pkg/object/command/sparkeks/entrypoint.go | 2 -- internal/pkg/object/command/sparkeks/sparkeks.go | 7 +------ 2 files changed, 1 insertion(+), 8 deletions(-) diff --git a/internal/pkg/object/command/sparkeks/entrypoint.go b/internal/pkg/object/command/sparkeks/entrypoint.go index 7097f7b1..e274d641 100644 --- a/internal/pkg/object/command/sparkeks/entrypoint.go +++ b/internal/pkg/object/command/sparkeks/entrypoint.go @@ -75,8 +75,6 @@ func (s sqlWrapperEntrypointStrategy) apply(spec *v1beta2.SparkApplicationSpec) spec.Arguments = buildArguments(s.arguments, s.appName, s.queryURI, s.user, s.resultURI, s.returnResult) } -// --- Extension-based strategy selection --- - // entrypointFactory builds the entrypoint strategy for a job from its execution context. type entrypointFactory func(execCtx *executionContext) entrypointStrategy diff --git a/internal/pkg/object/command/sparkeks/sparkeks.go b/internal/pkg/object/command/sparkeks/sparkeks.go index a9f41c80..5e3abdfe 100644 --- a/internal/pkg/object/command/sparkeks/sparkeks.go +++ b/internal/pkg/object/command/sparkeks/sparkeks.go @@ -52,9 +52,6 @@ const ( defaultSparkAppAPIVersion = "sparkoperator.k8s.io/v1beta2" defaultSparkAppKind = "SparkApplication" - // defaultApplicationType is applied when neither the loaded template nor the JAR branch - // set Spec.Type, so the CRD's required enum field is never submitted empty (see Gap 4). - // JAR application types (jarApplicationTypes / resolveJarApplicationType) live in entrypoint.go. defaultApplicationType = v1beta2.SparkApplicationTypePython queriesPath = "queries" @@ -740,9 +737,7 @@ func applySparkOperatorConfig(execCtx *executionContext) { sparkApp.ObjectMeta.Name = execCtx.appName sparkApp.ObjectMeta.Namespace = execCtx.commandContext.KubeNamespace - // Set the main application file (common to both entrypoint styles), then delegate the - // entrypoint-specific spec (Type / MainClass / Arguments) to a strategy: JAR/--class vs the - // default SQL wrapper (see entrypointStrategy above). + // Set main application file and arguments if execCtx.commandContext.WrapperURI != "" { s3aWrapperURI := updateS3ToS3aURI(execCtx.commandContext.WrapperURI) sparkApp.Spec.MainApplicationFile = &s3aWrapperURI From d1816874ec9c894e94de1cc3fb0560b5c7337893 Mon Sep 17 00:00:00 2001 From: Prasad Lohakpure Date: Fri, 31 Jul 2026 14:13:10 +0530 Subject: [PATCH 3/3] Added test cases, handling for empty entry_point --- .../pkg/object/command/sparkeks/entrypoint.go | 13 +- .../command/sparkeks/entrypoint_test.go | 154 ++++++++++++++++++ .../pkg/object/command/sparkeks/sparkeks.go | 13 +- 3 files changed, 173 insertions(+), 7 deletions(-) create mode 100644 internal/pkg/object/command/sparkeks/entrypoint_test.go diff --git a/internal/pkg/object/command/sparkeks/entrypoint.go b/internal/pkg/object/command/sparkeks/entrypoint.go index e274d641..d708e11a 100644 --- a/internal/pkg/object/command/sparkeks/entrypoint.go +++ b/internal/pkg/object/command/sparkeks/entrypoint.go @@ -29,7 +29,7 @@ func resolveJarApplicationType(applicationType string) v1beta2.SparkApplicationT } type entrypointStrategy interface { - apply(spec *v1beta2.SparkApplicationSpec) + apply(spec *v1beta2.SparkApplicationSpec) error } func buildArguments(override []string, appName, queryURI, user, resultURI string, returnResult bool) []string { @@ -54,12 +54,16 @@ type jarEntrypointStrategy struct { arguments []string } -func (s jarEntrypointStrategy) apply(spec *v1beta2.SparkApplicationSpec) { - mainClass := s.mainClass +func (s jarEntrypointStrategy) apply(spec *v1beta2.SparkApplicationSpec) error { + mainClass := strings.TrimSpace(s.mainClass) + if mainClass == "" { + return ErrMissingEntryPoint + } spec.Type = s.appType spec.MainClass = &mainClass spec.Arguments = buildArguments(s.arguments, s.appName, s.queryURI, s.user, s.resultURI, s.returnResult) + return nil } type sqlWrapperEntrypointStrategy struct { @@ -71,8 +75,9 @@ type sqlWrapperEntrypointStrategy struct { arguments []string } -func (s sqlWrapperEntrypointStrategy) apply(spec *v1beta2.SparkApplicationSpec) { +func (s sqlWrapperEntrypointStrategy) apply(spec *v1beta2.SparkApplicationSpec) error { spec.Arguments = buildArguments(s.arguments, s.appName, s.queryURI, s.user, s.resultURI, s.returnResult) + return nil } // entrypointFactory builds the entrypoint strategy for a job from its execution context. diff --git a/internal/pkg/object/command/sparkeks/entrypoint_test.go b/internal/pkg/object/command/sparkeks/entrypoint_test.go new file mode 100644 index 00000000..7a24f19d --- /dev/null +++ b/internal/pkg/object/command/sparkeks/entrypoint_test.go @@ -0,0 +1,154 @@ +package sparkeks + +import ( + "reflect" + "testing" + + "github.com/kubeflow/spark-operator/v2/api/v1beta2" + + "github.com/patterninc/heimdall/pkg/object" + "github.com/patterninc/heimdall/pkg/object/job" +) + +func TestResolveJarApplicationType(t *testing.T) { + tests := []struct { + name string + applicationType string + expected v1beta2.SparkApplicationType + }{ + {name: "scala lowercase", applicationType: "scala", expected: v1beta2.SparkApplicationTypeScala}, + {name: "java lowercase", applicationType: "java", expected: v1beta2.SparkApplicationTypeJava}, + {name: "java mixed case", applicationType: "Java", expected: v1beta2.SparkApplicationTypeJava}, + {name: "scala uppercase", applicationType: "SCALA", expected: v1beta2.SparkApplicationTypeScala}, + {name: "java with surrounding whitespace", applicationType: " java ", expected: v1beta2.SparkApplicationTypeJava}, + {name: "empty defaults to scala", applicationType: "", expected: defaultJarApplicationType}, + {name: "whitespace only defaults to scala", applicationType: " ", expected: defaultJarApplicationType}, + {name: "unrecognized defaults to scala", applicationType: "python", expected: defaultJarApplicationType}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := resolveJarApplicationType(tt.applicationType); got != tt.expected { + t.Errorf("resolveJarApplicationType(%q) = %v, want %v", tt.applicationType, got, tt.expected) + } + }) + } +} +func TestNewEntrypointStrategy_Selection(t *testing.T) { + tests := []struct { + name string + wrapperURI string + expectedType reflect.Type + }{ + {name: "jar selects jar strategy", wrapperURI: "s3://bucket/app.jar", expectedType: reflect.TypeOf(jarEntrypointStrategy{})}, + {name: "py selects sql wrapper strategy", wrapperURI: "s3://bucket/wrapper.py", expectedType: reflect.TypeOf(sqlWrapperEntrypointStrategy{})}, + {name: "jar uppercase extension", wrapperURI: "s3://bucket/APP.JAR", expectedType: reflect.TypeOf(jarEntrypointStrategy{})}, + {name: "py mixed case selects sql strategy", wrapperURI: "s3://bucket/Wrapper.Py", expectedType: reflect.TypeOf(sqlWrapperEntrypointStrategy{})}, + {name: "unknown extension defaults to sql wrapper", wrapperURI: "s3://bucket/app.sh", expectedType: reflect.TypeOf(sqlWrapperEntrypointStrategy{})}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + execCtx := newTestExecutionContext(tt.wrapperURI, &jobContext{}, "alice") + strategy := newEntrypointStrategy(execCtx) + if got := reflect.TypeOf(strategy); got != tt.expectedType { + t.Errorf("newEntrypointStrategy(%q) type = %v, want %v", tt.wrapperURI, got, tt.expectedType) + } + }) + } +} + +func TestNewJarEntrypointStrategy_Apply(t *testing.T) { + jobCtx := &jobContext{ + ReturnResult: true, + Parameters: &jobParameters{ + EntryPoint: "com.example.Main", + ApplicationType: "java", + }, + } + execCtx := newTestExecutionContext("s3://bucket/app.jar", jobCtx, "alice") + + strategy := newEntrypointStrategy(execCtx) + spec := &v1beta2.SparkApplicationSpec{} + if err := strategy.apply(spec); err != nil { + t.Fatalf("apply() returned unexpected error: %v", err) + } + + if spec.Type != v1beta2.SparkApplicationTypeJava { + t.Errorf("spec.Type = %v, want %v", spec.Type, v1beta2.SparkApplicationTypeJava) + } + if spec.MainClass == nil || *spec.MainClass != "com.example.Main" { + t.Errorf("spec.MainClass = %v, want com.example.Main", spec.MainClass) + } + expectedArgs := []string{"spark-sql-job-test", "s3a://bucket/query.sql", "alice", "s3a://bucket/result"} + if !reflect.DeepEqual(spec.Arguments, expectedArgs) { + t.Errorf("spec.Arguments = %v, want %v", spec.Arguments, expectedArgs) + } +} + +func TestNewSQLWrapperEntrypointStrategy_Apply(t *testing.T) { + jobCtx := &jobContext{ReturnResult: false} + execCtx := newTestExecutionContext("s3://bucket/wrapper.py", jobCtx, "bob") + + strategy := newEntrypointStrategy(execCtx) + spec := &v1beta2.SparkApplicationSpec{} + if err := strategy.apply(spec); err != nil { + t.Fatalf("apply() returned unexpected error: %v", err) + } + + if spec.MainClass != nil { + t.Errorf("spec.MainClass = %v, want nil for sql wrapper", spec.MainClass) + } + expectedArgs := []string{"spark-sql-job-test", "s3a://bucket/query.sql", "bob"} + if !reflect.DeepEqual(spec.Arguments, expectedArgs) { + t.Errorf("spec.Arguments = %v, want %v", spec.Arguments, expectedArgs) + } +} + +func TestJarEntrypointStrategy_Apply_MissingEntryPoint(t *testing.T) { + tests := []struct { + name string + jobCtx *jobContext + }{ + {name: "nil parameters", jobCtx: &jobContext{}}, + {name: "empty entry_point", jobCtx: &jobContext{Parameters: &jobParameters{EntryPoint: ""}}}, + {name: "whitespace entry_point", jobCtx: &jobContext{Parameters: &jobParameters{EntryPoint: " "}}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + execCtx := newTestExecutionContext("s3://bucket/app.jar", tt.jobCtx, "alice") + strategy := newEntrypointStrategy(execCtx) + spec := &v1beta2.SparkApplicationSpec{} + err := strategy.apply(spec) + if err != ErrMissingEntryPoint { + t.Errorf("apply() error = %v, want %v", err, ErrMissingEntryPoint) + } + if spec.MainClass != nil { + t.Errorf("spec.MainClass = %v, want nil when entry_point is missing", spec.MainClass) + } + }) + } +} + +func TestSQLWrapperEntrypointStrategy_Apply_NoEntryPointRequired(t *testing.T) { + execCtx := newTestExecutionContext("s3://bucket/wrapper.py", &jobContext{}, "alice") + strategy := newEntrypointStrategy(execCtx) + spec := &v1beta2.SparkApplicationSpec{} + if err := strategy.apply(spec); err != nil { + t.Errorf("apply() returned unexpected error: %v", err) + } +} + +func newTestExecutionContext(wrapperURI string, jobCtx *jobContext, user string) *executionContext { + return &executionContext{ + job: &job.Job{ + Object: object.Object{User: user}, + }, + commandContext: &commandContext{WrapperURI: wrapperURI}, + jobContext: jobCtx, + appName: "spark-sql-job-test", + s3aQueryURI: "s3a://bucket/query.sql", + s3aResultURI: "s3a://bucket/result", + } +} diff --git a/internal/pkg/object/command/sparkeks/sparkeks.go b/internal/pkg/object/command/sparkeks/sparkeks.go index 5e3abdfe..89c48871 100644 --- a/internal/pkg/object/command/sparkeks/sparkeks.go +++ b/internal/pkg/object/command/sparkeks/sparkeks.go @@ -94,6 +94,7 @@ var ( ErrKubeConfig = fmt.Errorf("failed to configure Kubernetes client: ensure EKS cluster access is properly configured") ErrApplicationSpec = fmt.Errorf("failed to load or parse SparkApplication template") ErrSparkApplicationFile = fmt.Errorf("failed to read SparkApplication application template file: check file path and permissions") + ErrMissingEntryPoint = fmt.Errorf("entry_point is required for .jar wrapper_uri: set parameters.entry_point to the fully-qualified main class") ) type commandContext struct { @@ -728,7 +729,7 @@ func updateKubeConfig(ctx context.Context, execCtx *executionContext) (string, e } // applySparkOperatorConfig consolidates all Spark Operator configuration updates and overrides. -func applySparkOperatorConfig(execCtx *executionContext) { +func applySparkOperatorConfig(execCtx *executionContext) error { sparkApp := execCtx.sparkApp jobContext := execCtx.jobContext clusterContext := execCtx.clusterContext @@ -742,7 +743,9 @@ func applySparkOperatorConfig(execCtx *executionContext) { s3aWrapperURI := updateS3ToS3aURI(execCtx.commandContext.WrapperURI) sparkApp.Spec.MainApplicationFile = &s3aWrapperURI // Type / MainClass / Arguments are set by the entrypoint strategy. - newEntrypointStrategy(execCtx).apply(&sparkApp.Spec) + if err := newEntrypointStrategy(execCtx).apply(&sparkApp.Spec); err != nil { + return err + } } if sparkApp.Spec.SparkConf == nil { @@ -852,6 +855,8 @@ func applySparkOperatorConfig(execCtx *executionContext) { if sparkApp.Spec.Type == "" { sparkApp.Spec.Type = defaultApplicationType } + + return nil } // generateSparkApp generates the Spark application from the template. @@ -864,7 +869,9 @@ func generateSparkApp(execCtx *executionContext) error { execCtx.sparkApp = sparkApp // Apply all configurations and overrides from contexts - applySparkOperatorConfig(execCtx) + if err := applySparkOperatorConfig(execCtx); err != nil { + return err + } return nil }