FEAT: Support for custom entrypoint in sparkeks

internal/pkg/object/command/sparkeks/entrypoint.go

@@ -0,0 +1,134 @@
+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) error
+}
+
+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) 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 {
+	appName      string
+	queryURI     string
+	user         string
+	resultURI    string
+	returnResult bool
+	arguments    []string
+}
+
+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.
+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,
+	}
+}

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",
+	}
+}