You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2025/09/19 02:38:23 UTC
(spark-kubernetes-operator) branch main updated: [SPARK-53642] Improve `Javadoc` for interfaces, classes, methods, variables
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new a6cbad6 [SPARK-53642] Improve `Javadoc` for interfaces, classes, methods, variables
a6cbad6 is described below
commit a6cbad62fc57b1a21089c5bb5212386aacb39ef1
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Thu Sep 18 19:38:14 2025 -0700
[SPARK-53642] Improve `Javadoc` for interfaces, classes, methods, variables
### What changes were proposed in this pull request?
This PR aims to improve `Javadoc` for interfaces, classes, methods, and variables as a part of v0.5 preparation.
### Why are the changes needed?
Currently, we have comments partially. We had better be consistent as much as possible.
- Fill many javadoc comments.
- Fill missing `param` descriptions.
- Fix typos.
- Unify the description style like a sentence, e.g., adding a period at the end.
After making this is a baseline, we can add and revise more in the future at every releases.
### Does this PR introduce _any_ user-facing change?
No behavior change because this is a documentation only change.
### How was this patch tested?
Pass the CIs (including building `javadoc` successfully).
### Was this patch authored or co-authored using generative AI tooling?
Yes, I asked `gemini-2.5-flash` to fill the comments first and revised.
Closes #324 from dongjoon-hyun/SPARK-53642.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../k8s/operator/utils/ConfOptionDocGenerator.java | 1 +
.../apache/spark/k8s/operator/utils/DocTable.java | 1 +
.../org/apache/spark/k8s/operator/Constants.java | 102 +++++++++++++++
.../spark/k8s/operator/SparkApplication.java | 10 ++
.../apache/spark/k8s/operator/SparkCluster.java | 10 ++
.../k8s/operator/decorators/ResourceDecorator.java | 8 ++
.../apache/spark/k8s/operator/diff/Diffable.java | 5 +
.../k8s/operator/spec/ApplicationTolerations.java | 23 ++--
.../spark/k8s/operator/spec/DeploymentMode.java | 1 +
.../spark/k8s/operator/spec/RestartPolicy.java | 9 ++
.../operator/status/ApplicationStateSummary.java | 34 ++++-
.../k8s/operator/status/ApplicationStatus.java | 41 ++++--
.../spark/k8s/operator/status/AttemptInfo.java | 5 +
.../k8s/operator/status/BaseStateSummary.java | 12 +-
.../spark/k8s/operator/status/BaseStatus.java | 20 +++
.../k8s/operator/status/ClusterStateSummary.java | 25 +++-
.../spark/k8s/operator/status/ClusterStatus.java | 22 +++-
.../spark/k8s/operator/utils/ModelUtils.java | 47 ++++++-
.../apache/spark/k8s/operator/SparkOperator.java | 43 ++++++-
.../operator/client/KubernetesClientFactory.java | 13 ++
.../spark/k8s/operator/config/ConfigOption.java | 12 ++
.../k8s/operator/config/SparkOperatorConf.java | 140 +++++++++++++++++++++
.../operator/config/SparkOperatorConfManager.java | 22 ++++
.../config/SparkOperatorConfigMapReconciler.java | 16 +++
.../spark/k8s/operator/context/BaseContext.java | 10 ++
.../k8s/operator/context/SparkAppContext.java | 35 ++++++
.../k8s/operator/context/SparkClusterContext.java | 35 ++++++
.../k8s/operator/decorators/ClusterDecorator.java | 8 +-
.../k8s/operator/decorators/DriverDecorator.java | 8 +-
.../decorators/DriverResourceDecorator.java | 8 ++
.../k8s/operator/listeners/BaseStatusListener.java | 7 ++
.../k8s/operator/metrics/BaseOperatorSource.java | 36 ++++++
.../spark/k8s/operator/metrics/JVMMetricSet.java | 6 +
.../spark/k8s/operator/metrics/MetricsService.java | 8 ++
.../spark/k8s/operator/metrics/MetricsSystem.java | 23 ++++
.../k8s/operator/metrics/MetricsSystemFactory.java | 17 +++
.../metrics/PrometheusPullModelHandler.java | 84 +++++++++++++
.../metrics/SparkAppStatusRecorderSource.java | 18 +++
.../metrics/healthcheck/SentinelManager.java | 44 +++++++
.../source/KubernetesMetricsInterceptor.java | 24 ++++
.../metrics/source/OperatorJosdkMetrics.java | 76 +++++++++++
.../operator/metrics/source/OperatorJvmSource.java | 10 ++
.../spark/k8s/operator/probe/HealthProbe.java | 12 ++
.../spark/k8s/operator/probe/ProbeService.java | 9 ++
.../spark/k8s/operator/probe/ReadinessProbe.java | 16 +++
.../k8s/operator/reconciler/ReconcileProgress.java | 31 +++++
.../operator/reconciler/SparkAppReconciler.java | 36 +++++-
.../reconciler/SparkAppResourceSpecFactory.java | 55 +++++++-
.../reconciler/SparkClusterReconciler.java | 37 +++++-
.../SparkClusterResourceSpecFactory.java | 7 ++
.../observers/AppDriverReadyObserver.java | 9 ++
.../observers/AppDriverRunningObserver.java | 9 ++
.../observers/AppDriverStartObserver.java | 9 ++
.../observers/AppDriverTimeoutObserver.java | 9 +-
.../observers/BaseAppDriverObserver.java | 10 +-
.../observers/BaseSecondaryResourceObserver.java | 8 ++
.../reconciler/reconcilesteps/AppCleanUpStep.java | 61 ++++++++-
.../reconciler/reconcilesteps/AppInitStep.java | 13 ++
.../reconcilesteps/AppReconcileStep.java | 43 +++++++
.../reconcilesteps/AppResourceObserveStep.java | 8 ++
.../reconciler/reconcilesteps/AppRunningStep.java | 7 ++
.../reconcilesteps/AppUnknownStateStep.java | 7 ++
.../reconciler/reconcilesteps/AppValidateStep.java | 7 ++
.../reconciler/reconcilesteps/ClusterInitStep.java | 7 ++
.../reconcilesteps/ClusterReconcileStep.java | 35 ++++++
.../reconcilesteps/ClusterTerminatedStep.java | 8 ++
.../reconcilesteps/ClusterUnknownStateStep.java | 7 ++
.../reconcilesteps/ClusterValidateStep.java | 7 ++
.../k8s/operator/utils/ClassLoadingUtils.java | 4 +-
.../spark/k8s/operator/utils/LoggingUtils.java | 7 ++
.../apache/spark/k8s/operator/utils/PodPhase.java | 6 +
.../apache/spark/k8s/operator/utils/PodUtils.java | 37 ++++--
.../apache/spark/k8s/operator/utils/ProbeUtil.java | 7 ++
.../spark/k8s/operator/utils/ReconcilerUtils.java | 58 +++++++++
.../k8s/operator/utils/SparkAppStatusRecorder.java | 13 ++
.../k8s/operator/utils/SparkAppStatusUtils.java | 43 +++++++
.../operator/utils/SparkClusterStatusRecorder.java | 11 ++
.../k8s/operator/utils/SparkExceptionUtils.java | 12 ++
.../spark/k8s/operator/utils/StatusRecorder.java | 24 +++-
.../org/apache/spark/k8s/operator/utils/Utils.java | 91 +++++++++++++-
.../spark/k8s/operator/SparkAppDriverConf.java | 29 +++--
.../spark/k8s/operator/SparkAppResourceSpec.java | 28 +++++
.../k8s/operator/SparkAppSubmissionWorker.java | 64 +++++++---
.../k8s/operator/SparkClusterResourceSpec.java | 61 +++++++++
.../k8s/operator/SparkClusterSubmissionWorker.java | 7 ++
.../k8s/operator/utils/ConfigMapSpecUtils.java | 6 +
.../operator/utils/DriverServiceIngressUtils.java | 23 +++-
87 files changed, 1975 insertions(+), 102 deletions(-)
diff --git a/build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/ConfOptionDocGenerator.java b/build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/ConfOptionDocGenerator.java
index c0f2f49..b2c3c63 100644
--- a/build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/ConfOptionDocGenerator.java
+++ b/build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/ConfOptionDocGenerator.java
@@ -30,6 +30,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.spark.k8s.operator.config.ConfigOption;
import org.apache.spark.k8s.operator.config.SparkOperatorConf;
+/** Generates documentation for configuration options. */
@Slf4j
public class ConfOptionDocGenerator {
public static final String CONF_FILE_NAME = "config_properties.md";
diff --git a/build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/DocTable.java b/build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/DocTable.java
index ab51487..569f012 100644
--- a/build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/DocTable.java
+++ b/build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/DocTable.java
@@ -28,6 +28,7 @@ import lombok.Builder;
import lombok.Data;
import lombok.RequiredArgsConstructor;
+/** Represents a table for documentation generation. */
@Data
@RequiredArgsConstructor
@Builder
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
index 3f7fff2..fb7f617 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
@@ -22,79 +22,181 @@ package org.apache.spark.k8s.operator;
/** Constants used in the Spark Kubernetes Operator. */
@SuppressWarnings("PMD.DataClass")
public class Constants {
+ /** The API group for Spark K8s Operator CRD resources. */
public static final String API_GROUP = "spark.apache.org";
+
+ /** The API version for Spark K8s Operator CRD resources. */
public static final String API_VERSION = "v1";
+
+ /** The label for the Spark Application name. */
public static final String LABEL_SPARK_APPLICATION_NAME = "spark.operator/spark-app-name";
+
+ /** The label for the Spark Cluster name. */
public static final String LABEL_SPARK_CLUSTER_NAME = "spark.operator/spark-cluster-name";
+
+ /** The label for the Spark Operator name. */
public static final String LABEL_SPARK_OPERATOR_NAME = "spark.operator/name";
+
+ /** The label for a sentinel resource. */
public static final String LABEL_SENTINEL_RESOURCE = "spark.operator/sentinel";
+
+ /** The label for the resource name. */
public static final String LABEL_RESOURCE_NAME = "app.kubernetes.io/name";
+
+ /** The label for the component name. */
public static final String LABEL_COMPONENT_NAME = "app.kubernetes.io/component";
+
+ /** The label for the Spark role name. */
public static final String LABEL_SPARK_ROLE_NAME = "spark-role";
+
+ /** The value for the Spark driver role label. */
public static final String LABEL_SPARK_ROLE_DRIVER_VALUE = "driver";
+
+ /** The value for the Spark executor role label. */
public static final String LABEL_SPARK_ROLE_EXECUTOR_VALUE = "executor";
+
+ /** The value for the Spark cluster role label. */
public static final String LABEL_SPARK_ROLE_CLUSTER_VALUE = "cluster";
+
+ /** The value for the Spark master role label. */
public static final String LABEL_SPARK_ROLE_MASTER_VALUE = "master";
+
+ /** The value for the Spark worker role label. */
public static final String LABEL_SPARK_ROLE_WORKER_VALUE = "worker";
+
+ /** The label for the Spark version. */
public static final String LABEL_SPARK_VERSION_NAME = "spark-version";
+
+ /** A dummy field for sentinel resources. */
public static final String SENTINEL_RESOURCE_DUMMY_FIELD = "sentinel.dummy.number";
+ /** The property key for the driver Spark container name. */
public static final String DRIVER_SPARK_CONTAINER_PROP_KEY =
"spark.kubernetes.driver.podTemplateContainerName";
+
+ /** The property key for the driver Spark pod template file. */
public static final String DRIVER_SPARK_TEMPLATE_FILE_PROP_KEY =
"spark.kubernetes.driver.podTemplateFile";
+
+ /** The property key for the executor Spark pod template file. */
public static final String EXECUTOR_SPARK_TEMPLATE_FILE_PROP_KEY =
"spark.kubernetes.executor.podTemplateFile";
// Default state messages
+ /** Message indicating that the driver has been requested from the resource scheduler. */
public static final String DRIVER_REQUESTED_MESSAGE = "Requested driver from resource scheduler.";
+
+ /** Message indicating that the Spark application completed successfully. */
public static final String DRIVER_COMPLETED_MESSAGE = "Spark application completed successfully.";
+
+ /**
+ * Message indicating that the driver container terminated before SparkContext/SparkSession
+ * initialization.
+ */
public static final String DRIVER_TERMINATED_BEFORE_INITIALIZATION_MESSAGE =
"Driver container is terminated without SparkContext / SparkSession initialization.";
+
+ /** Message indicating that the driver has failed init container(s). */
public static final String DRIVER_FAILED_INIT_CONTAINERS_MESSAGE =
"Driver has failed init container(s). Refer last observed status for details.";
+
+ /** Message indicating that the driver has one or more failed critical container(s). */
public static final String DRIVER_FAILED_MESSAGE =
"Driver has one or more failed critical container(s), refer last observed status for "
+ "details.";
+
+ /** Message indicating that the driver's critical container(s) exited with 0. */
public static final String DRIVER_SUCCEEDED_MESSAGE =
"Driver has critical container(s) exited with 0.";
+
+ /** Message indicating that the driver's critical container(s) restarted unexpectedly. */
public static final String DRIVER_RESTARTED_MESSAGE =
"Driver has one or more critical container(s) restarted unexpectedly, refer last "
+ "observed status for details.";
+
+ /** Message indicating that the Spark application has been shut down as requested. */
public static final String APP_CANCELLED_MESSAGE =
"Spark application has been shutdown as requested.";
+
+ /**
+ * Message indicating that Spark application resources were released after exceeding the
+ * configured retain duration.
+ */
public static final String APP_EXCEEDED_RETAIN_DURATION_MESSAGE =
"Spark application resources released after exceeding the configured retain duration.";
+
+ /** Message indicating that the driver was unexpectedly removed. */
public static final String DRIVER_UNEXPECTED_REMOVED_MESSAGE =
"Driver removed. This could caused by 'exit' called in driver process with non-zero "
+ "code, involuntary disruptions or unintentional destroy behavior, check "
+ "Kubernetes events for more details.";
+
+ /**
+ * Message indicating that the driver has not responded to the initial health check request within
+ * the allotted start-up time.
+ */
public static final String DRIVER_LAUNCH_TIMEOUT_MESSAGE =
"The driver has not responded to the initial health check request within the "
+ "allotted start-up time. This can be configured by setting "
+ ".spec.applicationTolerations.applicationTimeoutConfig.";
+
+ /** Message indicating that the driver has started running. */
public static final String DRIVER_RUNNING_MESSAGE = "Driver has started running.";
+
+ /** Message indicating that the driver has reached a ready state. */
public static final String DRIVER_READY_MESSAGE = "Driver has reached ready state.";
+
+ /** Message indicating that the Spark application has been created on the Kubernetes Cluster. */
public static final String SUBMITTED_STATE_MESSAGE =
"Spark application has been created on Kubernetes Cluster.";
+
+ /** Message indicating that the application status cannot be processed. */
public static final String UNKNOWN_STATE_MESSAGE = "Cannot process application status.";
+
+ /** Message indicating that the maximum number of restart attempts has been exceeded. */
public static final String EXCEED_MAX_RETRY_ATTEMPT_MESSAGE =
"The maximum number of restart attempts (%d) has been exceeded.";
+
+ /** Message indicating a failure to request the driver from the scheduler backend. */
public static final String SCHEDULE_FAILURE_MESSAGE =
"Failed to request driver from scheduler backend.";
+
+ /** Message indicating that the application is running healthy. */
public static final String RUNNING_HEALTHY_MESSAGE = "Application is running healthy.";
+
+ /**
+ * Message indicating that the application is running with less than the minimal number of
+ * requested initial executors.
+ */
public static final String INITIALIZED_WITH_BELOW_THRESHOLD_EXECUTORS_MESSAGE =
"The application is running with less than minimal number of requested initial executors.";
+
+ /**
+ * Message indicating that the Spark application is running with less than the minimal number of
+ * requested executors.
+ */
public static final String RUNNING_WITH_BELOW_THRESHOLD_EXECUTORS_MESSAGE =
"The Spark application is running with less than minimal number of requested executors.";
+
+ /**
+ * Message indicating that the Spark application failed to get enough executors in the given time
+ * threshold.
+ */
public static final String EXECUTOR_LAUNCH_TIMEOUT_MESSAGE =
"The Spark application failed to get enough executors in the given time threshold.";
// Spark Cluster Messages
+ /** Message indicating a failure to request the Spark cluster from the scheduler backend. */
public static final String CLUSTER_SCHEDULE_FAILURE_MESSAGE =
"Failed to request Spark cluster from scheduler backend.";
+
+ /** Message indicating that the Spark cluster has been submitted to the Kubernetes Cluster. */
public static final String CLUSTER_SUBMITTED_STATE_MESSAGE =
"Spark cluster has been submitted to Kubernetes Cluster.";
+
+ /** Message indicating that the cluster has reached a ready state. */
public static final String CLUSTER_READY_MESSAGE = "Cluster has reached ready state.";
+
+ /** Message indicating that the cluster status cannot be processed. */
public static final String UNKNOWN_CLUSTER_STATE_MESSAGE = "Cannot process cluster status.";
}
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/SparkApplication.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/SparkApplication.java
index a95241d..4b197ac 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/SparkApplication.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/SparkApplication.java
@@ -46,11 +46,21 @@ public class SparkApplication
ApplicationState,
ApplicationSpec,
ApplicationStatus> {
+ /**
+ * Initializes and returns a new ApplicationStatus object.
+ *
+ * @return A new ApplicationStatus instance.
+ */
@Override
public ApplicationStatus initStatus() {
return new ApplicationStatus();
}
+ /**
+ * Initializes and returns a new ApplicationSpec object.
+ *
+ * @return A new ApplicationSpec instance.
+ */
@Override
public ApplicationSpec initSpec() {
return new ApplicationSpec();
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/SparkCluster.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/SparkCluster.java
index 2d0ed8d..3981c37 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/SparkCluster.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/SparkCluster.java
@@ -42,11 +42,21 @@ import org.apache.spark.k8s.operator.status.ClusterStatus;
public class SparkCluster
extends BaseResource<
ClusterStateSummary, ClusterAttemptSummary, ClusterState, ClusterSpec, ClusterStatus> {
+ /**
+ * Initializes and returns a new ClusterStatus object.
+ *
+ * @return A new ClusterStatus instance.
+ */
@Override
public ClusterStatus initStatus() {
return new ClusterStatus();
}
+ /**
+ * Initializes and returns a new ClusterSpec object.
+ *
+ * @return A new ClusterSpec instance.
+ */
@Override
public ClusterSpec initSpec() {
return new ClusterSpec();
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/decorators/ResourceDecorator.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/decorators/ResourceDecorator.java
index 8c7a3bf..2029650 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/decorators/ResourceDecorator.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/decorators/ResourceDecorator.java
@@ -21,6 +21,14 @@ package org.apache.spark.k8s.operator.decorators;
import io.fabric8.kubernetes.api.model.HasMetadata;
+/** Interface for decorating Kubernetes resources. */
public interface ResourceDecorator {
+ /**
+ * Decorates a Kubernetes resource.
+ *
+ * @param resource The resource to decorate.
+ * @param <T> The type of the resource, extending HasMetadata.
+ * @return The decorated resource.
+ */
<T extends HasMetadata> T decorate(T resource);
}
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/diff/Diffable.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/diff/Diffable.java
index f26a96a..e9b90b4 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/diff/Diffable.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/diff/Diffable.java
@@ -19,4 +19,9 @@
package org.apache.spark.k8s.operator.diff;
+/**
+ * Represents an entity that can be compared for differences.
+ *
+ * @param <T> the type of the entity to compare with
+ */
public interface Diffable<T> {}
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationTolerations.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationTolerations.java
index fb2adc4..77b6539 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationTolerations.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationTolerations.java
@@ -70,6 +70,10 @@ public class ApplicationTolerations {
protected Long ttlAfterStopMillis = -1L;
/**
+ * Computes the effective resource retention duration in milliseconds. This is the smaller of
+ * `resourceRetainDurationMillis` or `ttlAfterStopMillis` if both are non-negative. If only one is
+ * non-negative, that value is used. If both are negative, -1L is returned.
+ *
* @return The effective retain duration for secondary resources, which would be the smaller value
* of `resourceRetainDurationMillis` or `ttlAfterStopMillis`, if they are set to non-negative
* value. Return -1 if none of them are set.
@@ -89,11 +93,12 @@ public class ApplicationTolerations {
/**
* Check whether a terminated application has exceeded the resource retain duration at the
- * provided instant
+ * provided instant.
*
- * @param lastObservedState last observed state of the application
- * @return true if the app has terminated and resource retain duration is configured to a positive
- * value and the app is not within retain duration; false otherwise.
+ * @param lastObservedState The last observed state of the application.
+ * @param instant The instant to check against.
+ * @return True if the app has terminated, has a positive retain duration configured, and has
+ * exceeded that duration; false otherwise.
*/
public boolean exceedRetainDurationAtInstant(
ApplicationState lastObservedState, Instant instant) {
@@ -106,10 +111,10 @@ public class ApplicationTolerations {
}
/**
- * Indicates whether the reconciler need to perform retain duration check
+ * Indicates whether the reconciler needs to perform a retain duration check.
*
- * @return true if `resourceRetainDurationMillis` or `ttlAfterStopMillis` is set to non-negative
- * value
+ * @return True if `resourceRetainDurationMillis` or `ttlAfterStopMillis` is set to a non-negative
+ * value, false otherwise.
*/
@JsonIgnore
public boolean isRetainDurationEnabled() {
@@ -117,9 +122,9 @@ public class ApplicationTolerations {
}
/**
- * Indicates whether the reconciler need to perform ttl check
+ * Indicates whether the reconciler needs to perform a TTL (Time-To-Live) check.
*
- * @return true if `ttlAfterStopMillis` is set to non-negative value
+ * @return True if `ttlAfterStopMillis` is set to a non-negative value, false otherwise.
*/
@JsonIgnore
public boolean isTTLEnabled() {
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/DeploymentMode.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/DeploymentMode.java
index 47e6b3c..bf6a298 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/DeploymentMode.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/DeploymentMode.java
@@ -19,6 +19,7 @@
package org.apache.spark.k8s.operator.spec;
+/** Represents the deployment mode for a Spark application. */
public enum DeploymentMode {
ClusterMode,
ClientMode
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/RestartPolicy.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/RestartPolicy.java
index 8eac7ff..651f5c2 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/RestartPolicy.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/RestartPolicy.java
@@ -21,12 +21,21 @@ package org.apache.spark.k8s.operator.spec;
import org.apache.spark.k8s.operator.status.BaseStateSummary;
+/** Defines the restart policy for a Spark application. */
public enum RestartPolicy {
Always,
Never,
OnFailure,
OnInfrastructureFailure;
+ /**
+ * Determines if a restart should be attempted based on the restart policy and the current state
+ * summary.
+ *
+ * @param policy The RestartPolicy to evaluate.
+ * @param stateSummary The current BaseStateSummary of the application.
+ * @return True if a restart should be attempted, false otherwise.
+ */
public static boolean attemptRestartOnState(
final RestartPolicy policy, final BaseStateSummary stateSummary) {
return switch (policy) {
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java
index fc5fed1..77a7700 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java
@@ -111,39 +111,63 @@ public enum ApplicationStateSummary implements BaseStateSummary {
Failed,
DriverReadyTimedOut);
+ /**
+ * Checks if the application is in an initializing state.
+ *
+ * @return True if the state is Submitted or ScheduledToRestart, false otherwise.
+ */
public boolean isInitializing() {
return Submitted.equals(this) || ScheduledToRestart.equals(this);
}
+ /**
+ * Checks if the application is in a starting state.
+ *
+ * @return True if the state is past ScheduledToRestart but before RunningHealthy, false
+ * otherwise.
+ */
public boolean isStarting() {
return ScheduledToRestart.ordinal() < this.ordinal()
&& RunningHealthy.ordinal() > this.ordinal();
}
/**
- * A state is 'terminated' if and only if no further actions are needed to reconcile it
+ * Checks if the application is in the `terminated` state which means no further actions are
+ * needed to reconcile it.
*
- * @return true if the state indicates app has terminated
+ * @return True if the state indicates the app has terminated (ResourceReleased or
+ * TerminatedWithoutReleaseResources), false otherwise.
*/
public boolean isTerminated() {
return ResourceReleased.equals(this) || TerminatedWithoutReleaseResources.equals(this);
}
/**
- * When state is 'stopping', operator releases its resources based on retain policy, and perform
- * retry based on retry policy
+ * Checks if the application is in the `stopping` state. The operator releases its resources based
+ * on retain policy and perform retry based on retry policy.
*
- * @return true if an app is stopping
+ * @return True if the state is past RunningWithBelowThresholdExecutors but not yet terminated,
+ * false otherwise.
*/
public boolean isStopping() {
return RunningWithBelowThresholdExecutors.ordinal() < this.ordinal() && !isTerminated();
}
+ /**
+ * Checks if the application is in a failure state.
+ *
+ * @return True if the state is one of the defined failure states, false otherwise.
+ */
@Override
public boolean isFailure() {
return failures.contains(this);
}
+ /**
+ * Checks if the application is in an infrastructure failure state.
+ *
+ * @return True if the state is one of the defined infrastructure failure states, false otherwise.
+ */
@Override
public boolean isInfrastructureFailure() {
return infrastructureFailures.contains(this);
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStatus.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStatus.java
index 87bc31f..3b98eb6 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStatus.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStatus.java
@@ -42,10 +42,19 @@ import org.apache.spark.k8s.operator.spec.RestartPolicy;
public class ApplicationStatus
extends BaseStatus<ApplicationStateSummary, ApplicationState, ApplicationAttemptSummary> {
+ /** Constructs a new, empty ApplicationStatus. */
public ApplicationStatus() {
super(new ApplicationState(), new ApplicationAttemptSummary());
}
+ /**
+ * Constructs a new ApplicationStatus with the given state and history.
+ *
+ * @param currentState The current state of the application.
+ * @param stateTransitionHistory The history of state transitions.
+ * @param previousAttemptSummary Summary of the previous application attempt.
+ * @param currentAttemptSummary Summary of the current application attempt.
+ */
public ApplicationStatus(
ApplicationState currentState,
Map<Long, ApplicationState> stateTransitionHistory,
@@ -55,7 +64,10 @@ public class ApplicationStatus
}
/**
- * Create a new ApplicationStatus, set the given latest state as current and update state history
+ * Appends a new state to the application's status history and sets it as the current state.
+ *
+ * @param state The new ApplicationState to append.
+ * @return A new ApplicationStatus object with the updated state.
*/
public ApplicationStatus appendNewState(ApplicationState state) {
return new ApplicationStatus(
@@ -66,15 +78,13 @@ public class ApplicationStatus
}
/**
- * Create ApplicationStatus to be updated upon termination of current attempt, with respect to
- * current state and restart config.
+ * Creates an updated ApplicationStatus based on the termination or restart logic.
*
- * @param restartConfig restart config for the app
- * @param resourceRetainPolicy resourceRetainPolicy for the app
- * @param stateMessageOverride state message to be applied
- * @param trimStateTransitionHistory if enabled, operator would trim the state history, keeping
- * only previous and current attempt.
- * @return updated ApplicationStatus
+ * @param restartConfig The restart configuration for the application.
+ * @param resourceRetainPolicy The resource retention policy for the application.
+ * @param stateMessageOverride An optional message to override the default state message.
+ * @param trimStateTransitionHistory If true, the state transition history will be trimmed.
+ * @return An updated ApplicationStatus object.
*/
public ApplicationStatus terminateOrRestart(
final RestartConfig restartConfig,
@@ -154,6 +164,13 @@ public class ApplicationStatus
}
}
+ /**
+ * Creates an ApplicationState indicating that the application is terminated without releasing
+ * resources.
+ *
+ * @param stateMessageOverride An optional message to override the default state message.
+ * @return An ApplicationState object for termination without resource release.
+ */
private ApplicationState terminateAppWithoutReleaseResource(String stateMessageOverride) {
String stateMessage =
"Application is terminated without releasing resources as configured."
@@ -162,6 +179,12 @@ public class ApplicationStatus
ApplicationStateSummary.TerminatedWithoutReleaseResources, stateMessage);
}
+ /**
+ * Creates an updated state transition history with a new state appended.
+ *
+ * @param state The new ApplicationState to append.
+ * @return A Map representing the updated state transition history.
+ */
private Map<Long, ApplicationState> createUpdatedHistoryWithNewState(ApplicationState state) {
TreeMap<Long, ApplicationState> updatedHistory = new TreeMap<>(stateTransitionHistory);
updatedHistory.put(updatedHistory.lastKey() + 1L, state);
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/AttemptInfo.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/AttemptInfo.java
index 1222820..8f8e99a 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/AttemptInfo.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/AttemptInfo.java
@@ -39,6 +39,11 @@ import lombok.ToString;
public class AttemptInfo {
@Getter @Builder.Default protected final long id = 0L;
+ /**
+ * Creates a new AttemptInfo object representing the next attempt.
+ *
+ * @return A new AttemptInfo with an incremented ID.
+ */
public AttemptInfo createNextAttemptInfo() {
return new AttemptInfo(id + 1L);
}
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseStateSummary.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseStateSummary.java
index f398dc8..c8faed0 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseStateSummary.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseStateSummary.java
@@ -21,9 +21,17 @@ package org.apache.spark.k8s.operator.status;
/** State for Spark Custom Resource */
public interface BaseStateSummary {
- /** The CustomResource has failed */
+ /**
+ * Checks if the CustomResource has failed.
+ *
+ * @return True if the CustomResource has failed, false otherwise.
+ */
boolean isFailure();
- /** The CustomResource has failed for infrastructure reasons */
+ /**
+ * Checks if the CustomResource has failed due to infrastructure reasons.
+ *
+ * @return True if the CustomResource has failed for infrastructure reasons, false otherwise.
+ */
boolean isInfrastructureFailure();
}
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseStatus.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseStatus.java
index 455e76f..0bd618b 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseStatus.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseStatus.java
@@ -46,6 +46,12 @@ public class BaseStatus<S, STATE extends BaseState<S>, AS extends BaseAttemptSum
@Getter final AS previousAttemptSummary;
@Getter final AS currentAttemptSummary;
+ /**
+ * Constructs a new BaseStatus with an initial state and current attempt summary.
+ *
+ * @param initState The initial state of the resource.
+ * @param currentAttemptSummary The summary of the current attempt.
+ */
public BaseStatus(STATE initState, AS currentAttemptSummary) {
this.currentState = initState;
this.stateTransitionHistory = new TreeMap<>();
@@ -54,6 +60,15 @@ public class BaseStatus<S, STATE extends BaseState<S>, AS extends BaseAttemptSum
this.currentAttemptSummary = currentAttemptSummary;
}
+ /**
+ * Constructs a new BaseStatus with a given current state, state transition history, and attempt
+ * summaries.
+ *
+ * @param currentState The current state of the resource.
+ * @param stateTransitionHistory The history of state transitions.
+ * @param previousAttemptSummary The summary of the previous attempt.
+ * @param currentAttemptSummary The summary of the current attempt.
+ */
public BaseStatus(
STATE currentState,
Map<Long, STATE> stateTransitionHistory,
@@ -65,6 +80,11 @@ public class BaseStatus<S, STATE extends BaseState<S>, AS extends BaseAttemptSum
this.currentAttemptSummary = currentAttemptSummary;
}
+ /**
+ * Returns the ID of the current state in the state transition history.
+ *
+ * @return The ID of the current state.
+ */
protected long getCurrentStateId() {
return stateTransitionHistory.lastKey();
}
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ClusterStateSummary.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ClusterStateSummary.java
index b71a529..074a21a 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ClusterStateSummary.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ClusterStateSummary.java
@@ -35,28 +35,49 @@ public enum ClusterStateSummary implements BaseStateSummary {
/** all resources (pods, services .etc have been cleaned up) */
ResourceReleased;
+ /**
+ * Checks if the cluster is in an initializing state.
+ *
+ * @return True if the state is Submitted, false otherwise.
+ */
public boolean isInitializing() {
return Submitted.equals(this);
}
+ /**
+ * Checks if the cluster is in a starting state.
+ *
+ * @return True if the state is before RunningHealthy, false otherwise.
+ */
public boolean isStarting() {
return RunningHealthy.ordinal() > this.ordinal();
}
/**
- * A state is 'terminated' if and only if no further actions are needed to reconcile it.
+ * Checks if the cluster is in a terminated state.
*
- * @return true if the state indicates the cluster has terminated
+ * @return True if the state indicates the cluster has terminated (ResourceReleased), false
+ * otherwise.
*/
public boolean isTerminated() {
return ResourceReleased.equals(this);
}
+ /**
+ * Checks if the cluster is in a failure state.
+ *
+ * @return True if the state is SchedulingFailure or Failed, false otherwise.
+ */
@Override
public boolean isFailure() {
return SchedulingFailure.equals(this) || Failed.equals(this);
}
+ /**
+ * Checks if the cluster is in an infrastructure failure state.
+ *
+ * @return True if the state is SchedulingFailure, false otherwise.
+ */
@Override
public boolean isInfrastructureFailure() {
return SchedulingFailure.equals(this);
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ClusterStatus.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ClusterStatus.java
index 5cc97b3..4d45210 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ClusterStatus.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ClusterStatus.java
@@ -35,10 +35,19 @@ import lombok.ToString;
public class ClusterStatus
extends BaseStatus<ClusterStateSummary, ClusterState, ClusterAttemptSummary> {
+ /** Constructs a new, empty ClusterStatus. */
public ClusterStatus() {
super(new ClusterState(), new ClusterAttemptSummary());
}
+ /**
+ * Constructs a new ClusterStatus with the given state and history.
+ *
+ * @param currentState The current state of the cluster.
+ * @param stateTransitionHistory The history of state transitions.
+ * @param previousAttemptSummary Summary of the previous cluster attempt.
+ * @param currentAttemptSummary Summary of the current cluster attempt.
+ */
public ClusterStatus(
ClusterState currentState,
Map<Long, ClusterState> stateTransitionHistory,
@@ -47,7 +56,12 @@ public class ClusterStatus
super(currentState, stateTransitionHistory, previousAttemptSummary, currentAttemptSummary);
}
- /** Create a new ClusterStatus, set the given latest state as current and update state history */
+ /**
+ * Appends a new state to the cluster's status history and sets it as the current state.
+ *
+ * @param state The new ClusterState to append.
+ * @return A new ClusterStatus object with the updated state.
+ */
public ClusterStatus appendNewState(ClusterState state) {
return new ClusterStatus(
state,
@@ -56,6 +70,12 @@ public class ClusterStatus
currentAttemptSummary);
}
+ /**
+ * Creates an updated state transition history with a new state appended.
+ *
+ * @param state The new ClusterState to append.
+ * @return A Map representing the updated state transition history.
+ */
private Map<Long, ClusterState> createUpdatedHistoryWithNewState(ClusterState state) {
TreeMap<Long, ClusterState> updatedHistory = new TreeMap<>(stateTransitionHistory);
updatedHistory.put(updatedHistory.lastKey() + 1L, state);
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/utils/ModelUtils.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/utils/ModelUtils.java
index 5351594..24190cc 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/utils/ModelUtils.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/utils/ModelUtils.java
@@ -45,6 +45,12 @@ public final class ModelUtils {
private ModelUtils() {}
+ /**
+ * Converts a PodTemplateSpec to a Pod object.
+ *
+ * @param podTemplateSpec The PodTemplateSpec to convert.
+ * @return A new Pod object.
+ */
public static Pod getPodFromTemplateSpec(PodTemplateSpec podTemplateSpec) {
if (podTemplateSpec == null) {
return new PodBuilder().withNewMetadata().endMetadata().withNewSpec().endSpec().build();
@@ -57,9 +63,13 @@ public final class ModelUtils {
}
/**
- * Find the Spark main container(s) in driver pod. If `spark.kubernetes.driver
- * .podTemplateContainerName` is not set, all containers are considered as main container from
- * health monitoring perspective
+ * Finds the Spark main container(s) in the driver pod based on the application specification. If
+ * `spark.kubernetes.driver.podTemplateContainerName` is not set, all containers are considered as
+ * main containers.
+ *
+ * @param appSpec The ApplicationSpec of the Spark application.
+ * @param containerStatusList A list of ContainerStatus objects from the driver pod.
+ * @return A List of ContainerStatus objects representing the main containers.
*/
public static List<ContainerStatus> findDriverMainContainerStatus(
final ApplicationSpec appSpec, final List<ContainerStatus> containerStatusList) {
@@ -78,10 +88,10 @@ public final class ModelUtils {
}
/**
- * Build OwnerReference to the given resource
+ * Builds an OwnerReference to the given resource.
*
- * @param owner the owner
- * @return OwnerReference to be used for subresources
+ * @param owner The owner resource.
+ * @return An OwnerReference object to be used for subresources.
*/
public static OwnerReference buildOwnerReferenceTo(HasMetadata owner) {
return new OwnerReferenceBuilder()
@@ -93,6 +103,13 @@ public final class ModelUtils {
.build();
}
+ /**
+ * Converts a Kubernetes resource to its JSON string representation.
+ *
+ * @param resource The resource to convert.
+ * @param <T> The type of the resource, extending HasMetadata.
+ * @return A JSON string representation of the resource.
+ */
public static <T extends HasMetadata> String asJsonString(T resource) {
try {
return objectMapper.writeValueAsString(resource);
@@ -101,18 +118,36 @@ public final class ModelUtils {
}
}
+ /**
+ * Checks if overriding the driver template is enabled in the application specification.
+ *
+ * @param applicationSpec The ApplicationSpec to check.
+ * @return True if driver template override is enabled, false otherwise.
+ */
public static boolean overrideDriverTemplateEnabled(ApplicationSpec applicationSpec) {
return applicationSpec != null
&& applicationSpec.getDriverSpec() != null
&& applicationSpec.getDriverSpec().getPodTemplateSpec() != null;
}
+ /**
+ * Checks if overriding the executor template is enabled in the application specification.
+ *
+ * @param applicationSpec The ApplicationSpec to check.
+ * @return True if executor template override is enabled, false otherwise.
+ */
public static boolean overrideExecutorTemplateEnabled(ApplicationSpec applicationSpec) {
return applicationSpec != null
&& applicationSpec.getExecutorSpec() != null
&& applicationSpec.getExecutorSpec().getPodTemplateSpec() != null;
}
+ /**
+ * Retrieves the current attempt ID from a SparkApplication's status.
+ *
+ * @param app The SparkApplication to get the attempt ID from.
+ * @return The current attempt ID, or 0L if not available.
+ */
public static long getAttemptId(final SparkApplication app) {
long attemptId = 0L;
if (app.getStatus() != null && app.getStatus().getCurrentAttemptSummary() != null) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/SparkOperator.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/SparkOperator.java
index 1daffbc..4334567 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/SparkOperator.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/SparkOperator.java
@@ -78,6 +78,7 @@ public class SparkOperator {
private final MetricsService metricsService;
private final ExecutorService metricsResourcesSingleThreadPool;
+ /** Constructs a new SparkOperator, initializing all its components. */
public SparkOperator() {
this.metricsSystem = MetricsSystemFactory.createMetricsSystem();
this.client =
@@ -107,6 +108,11 @@ public class SparkOperator {
this.metricsService = new MetricsService(metricsSystem, metricsResourcesSingleThreadPool);
}
+ /**
+ * Registers the Spark Application and Spark Cluster reconcilers with the operator.
+ *
+ * @return The Operator instance with registered controllers.
+ */
protected Operator registerSparkOperator() {
Operator op = new Operator(this::overrideOperatorConfigs);
registeredSparkControllers.add(
@@ -122,6 +128,11 @@ public class SparkOperator {
return op;
}
+ /**
+ * Registers a monitor for dynamic configuration changes via ConfigMaps.
+ *
+ * @return The Operator instance for the config monitor.
+ */
protected Operator registerSparkOperatorConfMonitor() {
Operator op = new Operator(this::overrideConfigMonitorConfigs);
String operatorNamespace = SparkOperatorConf.OPERATOR_NAMESPACE.getValue();
@@ -141,6 +152,12 @@ public class SparkOperator {
return op;
}
+ /**
+ * Updates the set of namespaces that the operator is watching.
+ *
+ * @param namespaces The new set of namespaces to watch.
+ * @return True if the namespaces were updated, false otherwise.
+ */
protected boolean updateWatchingNamespaces(Set<String> namespaces) {
if (watchedNamespaces.equals(namespaces)) {
log.info("No watched namespace change detected");
@@ -167,6 +184,11 @@ public class SparkOperator {
return true;
}
+ /**
+ * Overrides the default configuration for the operator.
+ *
+ * @param overrider The ConfigurationServiceOverrider to apply changes to.
+ */
protected void overrideOperatorConfigs(ConfigurationServiceOverrider overrider) {
overrider.withKubernetesClient(client);
overrider.withStopOnInformerErrorDuringStartup(
@@ -193,6 +215,11 @@ public class SparkOperator {
overrider.withUseSSAToPatchPrimaryResource(false);
}
+ /**
+ * Overrides the configuration for the dynamic config monitor.
+ *
+ * @param overrider The ConfigurationServiceOverrider to apply changes to.
+ */
protected void overrideConfigMonitorConfigs(ConfigurationServiceOverrider overrider) {
overrider.withKubernetesClient(client);
overrider.withConcurrentReconciliationThreads(
@@ -205,6 +232,11 @@ public class SparkOperator {
overrider.withUseSSAToPatchPrimaryResource(false);
}
+ /**
+ * Overrides the default configuration for individual controllers.
+ *
+ * @param overrider The ControllerConfigurationOverrider to apply changes to.
+ */
protected void overrideControllerConfigs(ControllerConfigurationOverrider<?> overrider) {
if (watchedNamespaces.isEmpty()) {
log.info("Initializing operator watching at cluster level.");
@@ -216,6 +248,13 @@ public class SparkOperator {
overrider.withRetry(SparkOperatorConf.getOperatorRetry());
}
+ /**
+ * Returns a list of OkHttp interceptors for the Kubernetes client, including metrics interceptors
+ * if enabled.
+ *
+ * @param metricsSystem The MetricsSystem to register interceptors with.
+ * @return A List of Interceptor objects.
+ */
protected List<Interceptor> getClientInterceptors(MetricsSystem metricsSystem) {
List<Interceptor> clientInterceptors = new ArrayList<>();
if (SparkOperatorConf.KUBERNETES_CLIENT_METRICS_ENABLED.getValue()) {
@@ -227,9 +266,9 @@ public class SparkOperator {
}
/**
- * Bootstrap Spark Operator
+ * Main entry point for the Spark Operator application.
*
- * @param args not used - operator behavior are configured from SparkOperatorConf
+ * @param args Command line arguments (not used).
*/
public static void main(String[] args) {
SparkOperator sparkOperator = new SparkOperator();
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/client/KubernetesClientFactory.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/client/KubernetesClientFactory.java
index 5e103df..d725b46 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/client/KubernetesClientFactory.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/client/KubernetesClientFactory.java
@@ -32,10 +32,23 @@ public final class KubernetesClientFactory {
private KubernetesClientFactory() {}
+ /**
+ * Builds a KubernetesClient with the given interceptors.
+ *
+ * @param interceptors A list of OkHttp interceptors to add to the client.
+ * @return A new KubernetesClient instance.
+ */
public static KubernetesClient buildKubernetesClient(final List<Interceptor> interceptors) {
return buildKubernetesClient(interceptors, null);
}
+ /**
+ * Builds a KubernetesClient with the given interceptors and configuration.
+ *
+ * @param interceptors A list of OkHttp interceptors to add to the client.
+ * @param kubernetesClientConfig The Kubernetes client configuration.
+ * @return A new KubernetesClient instance.
+ */
public static KubernetesClient buildKubernetesClient(
final List<Interceptor> interceptors, final Config kubernetesClientConfig) {
return new KubernetesClientBuilder()
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/ConfigOption.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/ConfigOption.java
index 4f7b03d..d5cfa86 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/ConfigOption.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/ConfigOption.java
@@ -49,6 +49,11 @@ public class ConfigOption<T> {
@Getter private T defaultValue;
@Getter private Class<T> typeParameterClass;
+ /**
+ * Returns the resolved value of the config option.
+ *
+ * @return The resolved value.
+ */
public T getValue() {
T resolvedValue = resolveValue();
if (log.isDebugEnabled()) {
@@ -82,6 +87,13 @@ public class ConfigOption<T> {
}
}
+ /**
+ * Resolves a string value to a primitive type or String.
+ *
+ * @param clazz The class of the target type.
+ * @param value The string value to resolve.
+ * @return The resolved value as an Object.
+ */
public static Object resolveValueToPrimitiveType(Class<?> clazz, String value) {
if (Boolean.class == clazz || Boolean.TYPE == clazz) {
return Boolean.parseBoolean(value);
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
index b3b352e..1e3e949 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
@@ -33,6 +33,7 @@ import org.apache.spark.k8s.operator.utils.Utils;
@Slf4j
public final class SparkOperatorConf {
+ /** Name of the operator. */
public static final ConfigOption<String> OPERATOR_APP_NAME =
ConfigOption.<String>builder()
.key("spark.kubernetes.operator.name")
@@ -42,6 +43,7 @@ public final class SparkOperatorConf {
.defaultValue("spark-kubernetes-operator")
.build();
+ /** Namespace that operator is deployed within. */
public static final ConfigOption<String> OPERATOR_NAMESPACE =
ConfigOption.<String>builder()
.key("spark.kubernetes.operator.namespace")
@@ -51,6 +53,10 @@ public final class SparkOperatorConf {
.defaultValue("default")
.build();
+ /**
+ * Comma-separated list of namespaces that the operator would be watching for Spark resources. If
+ * set to '*', operator would watch all namespaces.
+ */
public static final ConfigOption<String> OPERATOR_WATCHED_NAMESPACES =
ConfigOption.<String>builder()
.key("spark.kubernetes.operator.watchedNamespaces")
@@ -62,6 +68,10 @@ public final class SparkOperatorConf {
.defaultValue("default")
.build();
+ /**
+ * Enable to indicate informer errors should stop operator startup. If disabled, operator startup
+ * will ignore recoverable errors, caused for example by RBAC issues and will retry periodically.
+ */
public static final ConfigOption<Boolean> TERMINATE_ON_INFORMER_FAILURE_ENABLED =
ConfigOption.<Boolean>builder()
.key("spark.kubernetes.operator.terminateOnInformerFailureEnabled")
@@ -75,6 +85,7 @@ public final class SparkOperatorConf {
.defaultValue(false)
.build();
+ /** Grace period for operator shutdown before reconciliation threads are killed. */
public static final ConfigOption<Integer> RECONCILER_TERMINATION_TIMEOUT_SECONDS =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.reconciler.terminationTimeoutSeconds")
@@ -85,6 +96,10 @@ public final class SparkOperatorConf {
.defaultValue(30)
.build();
+ /**
+ * Thread pool size for Spark Operator reconcilers. Unbounded pool would be used if set to
+ * non-positive number.
+ */
public static final ConfigOption<Integer> RECONCILER_PARALLELISM =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.reconciler.parallelism")
@@ -96,6 +111,10 @@ public final class SparkOperatorConf {
.defaultValue(50)
.build();
+ /**
+ * Timeout (in seconds) to for requests made to API server. This applies only to foreground
+ * requests.
+ */
public static final ConfigOption<Long> RECONCILER_FOREGROUND_REQUEST_TIMEOUT_SECONDS =
ConfigOption.<Long>builder()
.key("spark.kubernetes.operator.reconciler.foregroundRequestTimeoutSeconds")
@@ -107,6 +126,13 @@ public final class SparkOperatorConf {
.defaultValue(30L)
.build();
+ /**
+ * Interval (in seconds, non-negative) to reconcile Spark applications. Note that reconciliation
+ * is always expected to be triggered when app spec / status is updated. This interval controls
+ * the reconcile behavior of operator reconciliation even when there's no update on
+ * SparkApplication, e.g. to determine whether a hanging app needs to be proactively terminated.
+ * Thus this is recommended to set to above 2 minutes to avoid unnecessary no-op reconciliation.
+ */
public static final ConfigOption<Long> RECONCILER_INTERVAL_SECONDS =
ConfigOption.<Long>builder()
.key("spark.kubernetes.operator.reconciler.intervalSeconds")
@@ -123,6 +149,10 @@ public final class SparkOperatorConf {
.defaultValue(120L)
.build();
+ /**
+ * When enabled, operator would trim state transition history when a new attempt starts, keeping
+ * previous attempt summary only.
+ */
public static final ConfigOption<Boolean> TRIM_ATTEMPT_STATE_TRANSITION_HISTORY =
ConfigOption.<Boolean>builder()
.key("spark.kubernetes.operator.reconciler.trimStateTransitionHistoryEnabled")
@@ -134,6 +164,7 @@ public final class SparkOperatorConf {
.defaultValue(true)
.build();
+ /** Comma-separated names of SparkAppStatusListener class implementations */
public static final ConfigOption<String> SPARK_APP_STATUS_LISTENER_CLASS_NAMES =
ConfigOption.<String>builder()
.key("spark.kubernetes.operator.reconciler.appStatusListenerClassNames")
@@ -143,6 +174,7 @@ public final class SparkOperatorConf {
.defaultValue("")
.build();
+ /** Comma-separated names of SparkClusterStatusListener class implementations */
public static final ConfigOption<String> SPARK_CLUSTER_STATUS_LISTENER_CLASS_NAMES =
ConfigOption.<String>builder()
.key("spark.kubernetes.operator.reconciler.clusterStatusListenerClassNames")
@@ -152,6 +184,11 @@ public final class SparkOperatorConf {
.defaultValue("")
.build();
+ /**
+ * When enabled, operator would use config map as source of truth for config property override.
+ * The config map need to be created in spark.kubernetes.operator.namespace, and labeled with
+ * operator name.
+ */
public static final ConfigOption<Boolean> DYNAMIC_CONFIG_ENABLED =
ConfigOption.<Boolean>builder()
.key("spark.kubernetes.operator.dynamicConfig.enabled")
@@ -164,6 +201,7 @@ public final class SparkOperatorConf {
.defaultValue(false)
.build();
+ /** The selector str applied to dynamic config map. */
public static final ConfigOption<String> DYNAMIC_CONFIG_SELECTOR =
ConfigOption.<String>builder()
.key("spark.kubernetes.operator.dynamicConfig.selector")
@@ -173,6 +211,10 @@ public final class SparkOperatorConf {
.defaultValue(Utils.labelsAsStr(Utils.defaultOperatorConfigLabels()))
.build();
+ /**
+ * Parallelism for dynamic config reconciler. Unbounded pool would be used if set to non-positive
+ * number.
+ */
public static final ConfigOption<Integer> DYNAMIC_CONFIG_RECONCILER_PARALLELISM =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.dynamicConfig.reconcilerParallelism")
@@ -184,6 +226,7 @@ public final class SparkOperatorConf {
.defaultValue(1)
.build();
+ /** Operator rate limiter refresh period(in seconds) for each resource. */
public static final ConfigOption<Integer> RECONCILER_RATE_LIMITER_REFRESH_PERIOD_SECONDS =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.reconciler.rateLimiter.refreshPeriodSeconds")
@@ -193,6 +236,10 @@ public final class SparkOperatorConf {
.defaultValue(15)
.build();
+ /**
+ * Max number of reconcile loops triggered within the rate limiter refresh period for each
+ * resource. Setting the limit to non-positive disables the limiter.
+ */
public static final ConfigOption<Integer> RECONCILER_RATE_LIMITER_MAX_LOOP_FOR_PERIOD =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.reconciler.rateLimiter.maxLoopForPeriod")
@@ -205,6 +252,7 @@ public final class SparkOperatorConf {
.defaultValue(5)
.build();
+ /** Initial interval(in seconds) of retries on unhandled controller errors. */
public static final ConfigOption<Integer> RECONCILER_RETRY_INITIAL_INTERVAL_SECONDS =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.reconciler.retry.initialIntervalSeconds")
@@ -214,6 +262,10 @@ public final class SparkOperatorConf {
.defaultValue(5)
.build();
+ /**
+ * Interval multiplier of retries on unhandled controller errors. Setting this to 1 for linear
+ * retry.
+ */
public static final ConfigOption<Double> RECONCILER_RETRY_INTERVAL_MULTIPLIER =
ConfigOption.<Double>builder()
.key("spark.kubernetes.operator.reconciler.retry.intervalMultiplier")
@@ -225,6 +277,10 @@ public final class SparkOperatorConf {
.defaultValue(1.5)
.build();
+ /**
+ * Max interval(in seconds) of retries on unhandled controller errors. Set to non-positive for
+ * unlimited.
+ */
public static final ConfigOption<Integer> RECONCILER_RETRY_MAX_INTERVAL_SECONDS =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.reconciler.retry.maxIntervalSeconds")
@@ -236,6 +292,10 @@ public final class SparkOperatorConf {
.defaultValue(-1)
.build();
+ /**
+ * Max attempts of retries on unhandled controller errors. Setting this to non-positive value
+ * means no retry.
+ */
public static final ConfigOption<Integer> API_RETRY_MAX_ATTEMPTS =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.api.retryMaxAttempts")
@@ -247,6 +307,10 @@ public final class SparkOperatorConf {
.defaultValue(15)
.build();
+ /**
+ * Default time (in seconds) to wait till next request. This would be used if server does not set
+ * Retry-After in response. Setting this to non-positive number means immediate retry.
+ */
public static final ConfigOption<Long> API_RETRY_ATTEMPT_AFTER_SECONDS =
ConfigOption.<Long>builder()
.key("spark.kubernetes.operator.api.retryAttemptAfterSeconds")
@@ -259,6 +323,11 @@ public final class SparkOperatorConf {
.defaultValue(1L)
.build();
+ /**
+ * Maximal number of retry attempts of requests to k8s server for resource status update. This
+ * would be performed on top of k8s client spark.kubernetes.operator.retry.maxAttempts to overcome
+ * potential conflicting update on the same SparkApplication. This should be positive number.
+ */
public static final ConfigOption<Long> API_STATUS_PATCH_MAX_ATTEMPTS =
ConfigOption.<Long>builder()
.key("spark.kubernetes.operator.api.statusPatchMaxAttempts")
@@ -273,6 +342,11 @@ public final class SparkOperatorConf {
.defaultValue(3L)
.build();
+ /**
+ * Maximal number of retry attempts of requesting secondary resource for Spark application. This
+ * would be performed on top of k8s client spark.kubernetes.operator.retry.maxAttempts to overcome
+ * potential conflicting reconcile on the same SparkApplication. This should be positive number
+ */
public static final ConfigOption<Long> API_SECONDARY_RESOURCE_CREATE_MAX_ATTEMPTS =
ConfigOption.<Long>builder()
.key("spark.kubernetes.operator.api.secondaryResourceCreateMaxAttempts")
@@ -287,6 +361,9 @@ public final class SparkOperatorConf {
.defaultValue(3L)
.build();
+ /**
+ * When enabled, the josdk metrics will be added in metrics source and configured for operator.
+ */
public static final ConfigOption<Boolean> JOSDK_METRICS_ENABLED =
ConfigOption.<Boolean>builder()
.key("spark.kubernetes.operator.metrics.josdkMetricsEnabled")
@@ -298,6 +375,11 @@ public final class SparkOperatorConf {
.defaultValue(true)
.build();
+ /**
+ * Enable KubernetesClient metrics for measuring the HTTP traffic to the Kubernetes API Server.
+ * Since the metrics is collected via Okhttp interceptors, can be disabled when opt in customized
+ * interceptors.
+ */
public static final ConfigOption<Boolean> KUBERNETES_CLIENT_METRICS_ENABLED =
ConfigOption.<Boolean>builder()
.key("spark.kubernetes.operator.metrics.clientMetricsEnabled")
@@ -311,6 +393,10 @@ public final class SparkOperatorConf {
.defaultValue(true)
.build();
+ /**
+ * When enabled, additional metrics group by http response code group(1xx, 2xx, 3xx, 4xx, 5xx)
+ * received from API server will be added.
+ */
public static final ConfigOption<Boolean>
KUBERNETES_CLIENT_METRICS_GROUP_BY_RESPONSE_CODE_GROUP_ENABLED =
ConfigOption.<Boolean>builder()
@@ -325,6 +411,7 @@ public final class SparkOperatorConf {
.defaultValue(true)
.build();
+ /** The port used for checking metrics */
public static final ConfigOption<Integer> OPERATOR_METRICS_PORT =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.metrics.port")
@@ -334,6 +421,7 @@ public final class SparkOperatorConf {
.defaultValue(19090)
.build();
+ /** Whether or not to enable text-based format for Prometheus 2.0. */
public static final ConfigOption<Boolean> PROMETHEUS_TEXT_BASED_FORMAT_ENABLED =
ConfigOption.<Boolean>builder()
.key("spark.kubernetes.operator.metrics.prometheusTextBasedFormatEnabled")
@@ -346,6 +434,7 @@ public final class SparkOperatorConf {
.defaultValue(true)
.build();
+ /** Whether or not to enable automatic name sanitizing for all metrics. */
public static final ConfigOption<Boolean> SANITIZE_PROMETHEUS_METRICS_NAME_ENABLED =
ConfigOption.<Boolean>builder()
.key("spark.kubernetes.operator.metrics.sanitizePrometheusMetricsNameEnabled")
@@ -358,6 +447,7 @@ public final class SparkOperatorConf {
.defaultValue(true)
.build();
+ /** The port used for health/readiness check probe status. */
public static final ConfigOption<Integer> OPERATOR_PROBE_PORT =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.health.probePort")
@@ -367,6 +457,7 @@ public final class SparkOperatorConf {
.defaultValue(19091)
.build();
+ /** Size of executor service in Sentinel Managers to check the health of sentinel resources. */
public static final ConfigOption<Integer> SENTINEL_EXECUTOR_SERVICE_POOL_SIZE =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.health.sentinelExecutorPoolSize")
@@ -378,6 +469,7 @@ public final class SparkOperatorConf {
.enableDynamicOverride(false)
.build();
+ /** Allowed max time(seconds) between spec update and reconciliation for sentinel resources. */
public static final ConfigOption<Integer> SENTINEL_RESOURCE_RECONCILIATION_DELAY =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.health.sentinelResourceReconciliationDelaySeconds")
@@ -389,6 +481,11 @@ public final class SparkOperatorConf {
.defaultValue(60)
.build();
+ /**
+ * Enable leader election for the operator to allow running standby instances. When this is
+ * disabled, only one operator instance is expected to be up and running at any time (replica = 1)
+ * to avoid race condition.
+ */
public static final ConfigOption<Boolean> LEADER_ELECTION_ENABLED =
ConfigOption.<Boolean>builder()
.key("spark.kubernetes.operator.leaderElection.enabled")
@@ -401,6 +498,7 @@ public final class SparkOperatorConf {
.defaultValue(false)
.build();
+ /** Leader election lease name, must be unique for leases in the same namespace. */
public static final ConfigOption<String> LEADER_ELECTION_LEASE_NAME =
ConfigOption.<String>builder()
.key("spark.kubernetes.operator.leaderElection.leaseName")
@@ -411,6 +509,7 @@ public final class SparkOperatorConf {
.defaultValue("spark-operator-lease")
.build();
+ /** Leader election lease duration in seconds, non-negative. */
public static final ConfigOption<Integer> LEADER_ELECTION_LEASE_DURATION_SECONDS =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.leaderElection.leaseDurationSeconds")
@@ -420,6 +519,10 @@ public final class SparkOperatorConf {
.defaultValue(180)
.build();
+ /**
+ * Leader election renew deadline in seconds, non-negative. This needs to be smaller than the
+ * lease duration to allow current leader renew the lease before lease expires.
+ */
public static final ConfigOption<Integer> LEADER_ELECTION_RENEW_DEADLINE_SECONDS =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.leaderElection.renewDeadlineSeconds")
@@ -432,6 +535,7 @@ public final class SparkOperatorConf {
.defaultValue(120)
.build();
+ /** Leader election retry period in seconds, non-negative. */
public static final ConfigOption<Integer> LEADER_ELECTION_RETRY_PERIOD_SECONDS =
ConfigOption.<Integer>builder()
.key("spark.kubernetes.operator.leaderElection.retryPeriodSeconds")
@@ -443,6 +547,11 @@ public final class SparkOperatorConf {
private SparkOperatorConf() {}
+ /**
+ * Returns the LeaderElectionConfiguration based on the configured options.
+ *
+ * @return A LeaderElectionConfiguration object.
+ */
public static LeaderElectionConfiguration getLeaderElectionConfig() {
return new LeaderElectionConfiguration(
LEADER_ELECTION_LEASE_NAME.getValue(),
@@ -452,6 +561,11 @@ public final class SparkOperatorConf {
Duration.ofSeconds(ensurePositiveIntFor(LEADER_ELECTION_RETRY_PERIOD_SECONDS)));
}
+ /**
+ * Returns a GenericRetry object configured with the operator's retry settings.
+ *
+ * @return A GenericRetry object.
+ */
public static GenericRetry getOperatorRetry() {
GenericRetry genericRetry =
new GenericRetry()
@@ -470,20 +584,46 @@ public final class SparkOperatorConf {
return genericRetry;
}
+ /**
+ * Returns a RateLimiter configured with the operator's rate limiting settings.
+ *
+ * @return A RateLimiter object.
+ */
public static RateLimiter<?> getOperatorRateLimiter() {
return new LinearRateLimiter(
Duration.ofSeconds(ensureNonNegativeIntFor(RECONCILER_RATE_LIMITER_REFRESH_PERIOD_SECONDS)),
ensureNonNegativeIntFor(RECONCILER_RATE_LIMITER_MAX_LOOP_FOR_PERIOD));
}
+ /**
+ * Ensures that the integer value of a ConfigOption is non-negative.
+ *
+ * @param configOption The ConfigOption to check.
+ * @return The non-negative integer value.
+ */
private static int ensureNonNegativeIntFor(ConfigOption<Integer> configOption) {
return ensureValid(configOption.getValue(), configOption.getDescription(), 0, 0);
}
+ /**
+ * Ensures that the integer value of a ConfigOption is positive.
+ *
+ * @param configOption The ConfigOption to check.
+ * @return The positive integer value.
+ */
private static int ensurePositiveIntFor(ConfigOption<Integer> configOption) {
return ensureValid(configOption.getValue(), configOption.getDescription(), 1, 1);
}
+ /**
+ * Ensures that a given integer value is within a valid range.
+ *
+ * @param value The value to validate.
+ * @param description A description of the value for logging purposes.
+ * @param minValue The minimum allowed value (inclusive).
+ * @param defaultValue The default value to use if the provided value is invalid.
+ * @return The validated value, or the default value if invalid.
+ */
private static int ensureValid(int value, String description, int minValue, int defaultValue) {
if (value < minValue) {
if (defaultValue < minValue) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConfManager.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConfManager.java
index 13e87a9..5def621 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConfManager.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConfManager.java
@@ -59,6 +59,12 @@ public class SparkOperatorConfManager {
initialize();
}
+ /**
+ * Returns the current value for a given configuration key, considering dynamic overrides.
+ *
+ * @param key The configuration key.
+ * @return The resolved configuration value.
+ */
public String getValue(String key) {
synchronized (this) {
String currentValue = configOverrides.getProperty(key);
@@ -66,10 +72,21 @@ public class SparkOperatorConfManager {
}
}
+ /**
+ * Returns the initial value for a given configuration key, without considering dynamic overrides.
+ *
+ * @param key The configuration key.
+ * @return The initial configuration value.
+ */
public String getInitialValue(String key) {
return initialConfig.getProperty(key);
}
+ /**
+ * Refreshes the configuration overrides with new values from a map.
+ *
+ * @param updatedConfig A map containing the updated configuration properties.
+ */
public void refresh(Map<String, String> updatedConfig) {
synchronized (this) {
this.configOverrides = new Properties();
@@ -77,6 +94,11 @@ public class SparkOperatorConfManager {
}
}
+ /**
+ * Returns the properties related to metrics configuration.
+ *
+ * @return A Properties object containing metrics configuration.
+ */
public Properties getMetricsProperties() {
return metricsConfig;
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConfigMapReconciler.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConfigMapReconciler.java
index 1a282b3..184129d 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConfigMapReconciler.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConfigMapReconciler.java
@@ -43,6 +43,14 @@ public class SparkOperatorConfigMapReconciler implements Reconciler<ConfigMap> {
private final Function<Set<String>, Boolean> namespaceUpdater;
private final Function<Void, Set<String>> watchedNamespacesGetter;
+ /**
+ * Updates the error status of the ConfigMap reconciliation.
+ *
+ * @param resource The ConfigMap resource.
+ * @param context The reconciliation context.
+ * @param e The exception that occurred.
+ * @return An ErrorStatusUpdateControl indicating no status update.
+ */
@Override
public ErrorStatusUpdateControl<ConfigMap> updateErrorStatus(
ConfigMap resource, Context<ConfigMap> context, Exception e) {
@@ -50,6 +58,14 @@ public class SparkOperatorConfigMapReconciler implements Reconciler<ConfigMap> {
return ErrorStatusUpdateControl.noStatusUpdate();
}
+ /**
+ * Reconciles the ConfigMap resource, refreshing the Spark Operator configuration.
+ *
+ * @param resource The ConfigMap resource to reconcile.
+ * @param context The reconciliation context.
+ * @return An UpdateControl indicating no update.
+ * @throws Exception if an error occurs during reconciliation.
+ */
@Override
public UpdateControl<ConfigMap> reconcile(ConfigMap resource, Context<ConfigMap> context)
throws Exception {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/BaseContext.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/BaseContext.java
index 82b2347..3e201d4 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/BaseContext.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/BaseContext.java
@@ -29,7 +29,17 @@ import org.apache.spark.k8s.operator.BaseResource;
* @param <CR> The type of the custom resource.
*/
public abstract class BaseContext<CR extends BaseResource<?, ?, ?, ?, ?>> {
+ /**
+ * Returns the custom resource associated with this context.
+ *
+ * @return The custom resource.
+ */
public abstract CR getResource();
+ /**
+ * Returns the Kubernetes client associated with this context.
+ *
+ * @return The Kubernetes client.
+ */
public abstract KubernetesClient getClient();
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkAppContext.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkAppContext.java
index 7dba5a2..30d0e4a 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkAppContext.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkAppContext.java
@@ -53,6 +53,11 @@ public class SparkAppContext extends BaseContext<SparkApplication> {
/** secondaryResourceSpec is initialized in a lazy fashion - built upon the first attempt */
private SparkAppResourceSpec secondaryResourceSpec;
+ /**
+ * Returns the driver pod for the Spark application, if present.
+ *
+ * @return An Optional containing the driver Pod, or empty if not found.
+ */
public Optional<Pod> getDriverPod() {
return josdkContext
.getSecondaryResourcesAsStream(Pod.class)
@@ -65,6 +70,11 @@ public class SparkAppContext extends BaseContext<SparkApplication> {
.findAny();
}
+ /**
+ * Returns a set of executor pods for the Spark application.
+ *
+ * @return A Set of Pods representing the executors.
+ */
public Set<Pod> getExecutorsForApplication() {
return josdkContext
.getSecondaryResourcesAsStream(Pod.class)
@@ -88,24 +98,49 @@ public class SparkAppContext extends BaseContext<SparkApplication> {
}
}
+ /**
+ * Returns the SparkApplication resource associated with this context.
+ *
+ * @return The SparkApplication resource.
+ */
@Override
public SparkApplication getResource() {
return sparkApplication;
}
+ /**
+ * Returns the Kubernetes client from the JOSDK context.
+ *
+ * @return The KubernetesClient instance.
+ */
@Override
public KubernetesClient getClient() {
return josdkContext.getClient();
}
+ /**
+ * Returns a list of pre-driver resources specifications.
+ *
+ * @return A List of HasMetadata objects representing pre-driver resources.
+ */
public List<HasMetadata> getDriverPreResourcesSpec() {
return getSecondaryResourceSpec().getDriverPreResources();
}
+ /**
+ * Returns the driver pod specification.
+ *
+ * @return The Pod object representing the driver pod specification.
+ */
public Pod getDriverPodSpec() {
return getSecondaryResourceSpec().getConfiguredPod();
}
+ /**
+ * Returns a list of driver resources specifications.
+ *
+ * @return A List of HasMetadata objects representing driver resources.
+ */
public List<HasMetadata> getDriverResourcesSpec() {
return getSecondaryResourceSpec().getDriverResources();
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkClusterContext.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkClusterContext.java
index 3df94c8..836139c 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkClusterContext.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkClusterContext.java
@@ -56,31 +56,66 @@ public class SparkClusterContext extends BaseContext<SparkCluster> {
}
}
+ /**
+ * Returns the SparkCluster resource associated with this context.
+ *
+ * @return The SparkCluster resource.
+ */
@Override
public SparkCluster getResource() {
return sparkCluster;
}
+ /**
+ * Returns the specification for the master service.
+ *
+ * @return The Service object for the master.
+ */
public Service getMasterServiceSpec() {
return getSecondaryResourceSpec().getMasterService();
}
+ /**
+ * Returns the specification for the worker service.
+ *
+ * @return The Service object for the workers.
+ */
public Service getWorkerServiceSpec() {
return getSecondaryResourceSpec().getWorkerService();
}
+ /**
+ * Returns the specification for the master StatefulSet.
+ *
+ * @return The StatefulSet object for the master.
+ */
public StatefulSet getMasterStatefulSetSpec() {
return getSecondaryResourceSpec().getMasterStatefulSet();
}
+ /**
+ * Returns the specification for the worker StatefulSet.
+ *
+ * @return The StatefulSet object for the workers.
+ */
public StatefulSet getWorkerStatefulSetSpec() {
return getSecondaryResourceSpec().getWorkerStatefulSet();
}
+ /**
+ * Returns the specification for the HorizontalPodAutoscaler, if present.
+ *
+ * @return An Optional containing the HorizontalPodAutoscaler object.
+ */
public Optional<HorizontalPodAutoscaler> getHorizontalPodAutoscalerSpec() {
return getSecondaryResourceSpec().getHorizontalPodAutoscaler();
}
+ /**
+ * Returns the Kubernetes client from the JOSDK context.
+ *
+ * @return The KubernetesClient instance.
+ */
@Override
public KubernetesClient getClient() {
return josdkContext.getClient();
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/ClusterDecorator.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/ClusterDecorator.java
index 68ab96b..09d075b 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/ClusterDecorator.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/ClusterDecorator.java
@@ -35,7 +35,13 @@ public class ClusterDecorator implements ResourceDecorator {
private final SparkCluster cluster;
- /** Add labels and owner references to the cluster for all secondary resources */
+ /**
+ * Decorates a Kubernetes resource with owner references and labels from the SparkCluster.
+ *
+ * @param resource The resource to decorate.
+ * @param <T> The type of the resource, extending HasMetadata.
+ * @return The decorated resource.
+ */
@Override
public <T extends HasMetadata> T decorate(T resource) {
ObjectMeta metaData =
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverDecorator.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverDecorator.java
index 7cab1f5..719f9d9 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverDecorator.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverDecorator.java
@@ -38,7 +38,13 @@ public class DriverDecorator implements ResourceDecorator {
private final SparkApplication app;
- /** Add labels and owner references to the app for all secondary resources */
+ /**
+ * Decorates a Kubernetes resource with owner references and labels from the SparkApplication.
+ *
+ * @param resource The resource to decorate.
+ * @param <T> The type of the resource, extending HasMetadata.
+ * @return The decorated resource.
+ */
@Override
public <T extends HasMetadata> T decorate(T resource) {
ObjectMeta metaData =
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverResourceDecorator.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverResourceDecorator.java
index 95a6155..711a9a8 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverResourceDecorator.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/decorators/DriverResourceDecorator.java
@@ -45,6 +45,14 @@ import lombok.extern.slf4j.Slf4j;
public class DriverResourceDecorator implements ResourceDecorator {
private final Pod driverPod;
+ /**
+ * Decorates a Kubernetes resource by adding an owner reference to the driver pod. This ensures
+ * that secondary resources are garbage collected when the driver pod is deleted.
+ *
+ * @param resource The resource to decorate.
+ * @param <T> The type of the resource, extending HasMetadata.
+ * @return The decorated resource.
+ */
@Override
public <T extends HasMetadata> T decorate(T resource) {
boolean ownerReferenceExists = false;
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/listeners/BaseStatusListener.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/listeners/BaseStatusListener.java
index f3d12c5..6e7c0d3 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/listeners/BaseStatusListener.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/listeners/BaseStatusListener.java
@@ -29,5 +29,12 @@ import org.apache.spark.k8s.operator.status.BaseStatus;
*/
public abstract class BaseStatusListener<
STATUS extends BaseStatus<?, ?, ?>, CR extends BaseResource<?, ?, ?, ?, STATUS>> {
+ /**
+ * Called when the status of a resource changes.
+ *
+ * @param resource The custom resource whose status has changed.
+ * @param prevStatus The previous status of the resource.
+ * @param updatedStatus The updated status of the resource.
+ */
public abstract void listenStatus(CR resource, STATUS prevStatus, STATUS updatedStatus);
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/BaseOperatorSource.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/BaseOperatorSource.java
index 3953cf6..3540131 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/BaseOperatorSource.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/BaseOperatorSource.java
@@ -29,6 +29,7 @@ import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;
import lombok.RequiredArgsConstructor;
+/** Base class for operator metric sources. */
@RequiredArgsConstructor
public class BaseOperatorSource {
protected final MetricRegistry metricRegistry;
@@ -37,6 +38,13 @@ public class BaseOperatorSource {
protected final Map<String, Gauge<?>> gauges = new ConcurrentHashMap<>();
protected final Map<String, Timer> timers = new ConcurrentHashMap<>();
+ /**
+ * Retrieves or creates a Histogram metric.
+ *
+ * @param metricNamePrefix The prefix for the metric name.
+ * @param names Additional names to form the full metric name.
+ * @return A Histogram instance.
+ */
protected Histogram getHistogram(String metricNamePrefix, String... names) {
Histogram histogram;
String metricName = MetricRegistry.name(metricNamePrefix, names).toLowerCase();
@@ -49,6 +57,13 @@ public class BaseOperatorSource {
return histogram;
}
+ /**
+ * Retrieves or creates a Counter metric.
+ *
+ * @param metricNamePrefix The prefix for the metric name.
+ * @param names Additional names to form the full metric name.
+ * @return A Counter instance.
+ */
protected Counter getCounter(String metricNamePrefix, String... names) {
Counter counter;
String metricName = MetricRegistry.name(metricNamePrefix, names).toLowerCase();
@@ -61,6 +76,14 @@ public class BaseOperatorSource {
return counter;
}
+ /**
+ * Retrieves or creates a Gauge metric.
+ *
+ * @param defaultGauge The default gauge to use if one doesn't exist.
+ * @param metricNamePrefix The prefix for the metric name.
+ * @param names Additional names to form the full metric name.
+ * @return A Gauge instance.
+ */
protected Gauge<?> getGauge(Gauge<?> defaultGauge, String metricNamePrefix, String... names) {
Gauge<?> gauge;
String metricName = MetricRegistry.name(metricNamePrefix, names).toLowerCase();
@@ -73,6 +96,13 @@ public class BaseOperatorSource {
return gauge;
}
+ /**
+ * Retrieves or creates a Timer metric.
+ *
+ * @param metricNamePrefix The prefix for the metric name.
+ * @param names Additional names to form the full metric name.
+ * @return A Timer instance.
+ */
protected Timer getTimer(String metricNamePrefix, String... names) {
Timer timer;
String metricName = MetricRegistry.name(metricNamePrefix, names).toLowerCase();
@@ -85,6 +115,12 @@ public class BaseOperatorSource {
return timer;
}
+ /**
+ * Returns the simple class name as the metric name prefix.
+ *
+ * @param klass The class to get the simple name from.
+ * @return The simple class name.
+ */
protected String getMetricNamePrefix(Class<?> klass) {
return klass.getSimpleName();
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/JVMMetricSet.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/JVMMetricSet.java
index 8a4a27a..745042d 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/JVMMetricSet.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/JVMMetricSet.java
@@ -47,6 +47,7 @@ public class JVMMetricSet implements MetricSet {
private final MemoryUsageGaugeSet memoryUsageGaugeSet;
private final ThreadStatesGaugeSet threadStatesGaugeSet;
+ /** Constructs a new JVMMetricSet, initializing all underlying JVM metric sets. */
public JVMMetricSet() {
bufferPoolMetricSet = new BufferPoolMetricSet(ManagementFactory.getPlatformMBeanServer());
fileDescriptorRatioGauge = new FileDescriptorRatioGauge();
@@ -55,6 +56,11 @@ public class JVMMetricSet implements MetricSet {
threadStatesGaugeSet = new ThreadStatesGaugeSet();
}
+ /**
+ * Returns a map of all JVM metrics collected by this MetricSet.
+ *
+ * @return A Map where keys are metric names and values are Metric objects.
+ */
@Override
public Map<String, Metric> getMetrics() {
final Map<String, Metric> jvmMetrics = new HashMap<>();
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/MetricsService.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/MetricsService.java
index ea5e6bd..99e9ced 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/MetricsService.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/MetricsService.java
@@ -34,6 +34,12 @@ public class MetricsService {
HttpServer server;
final MetricsSystem metricsSystem;
+ /**
+ * Constructs a new MetricsService.
+ *
+ * @param metricsSystem The MetricsSystem to expose metrics from.
+ * @param executor The Executor to use for the HTTP server.
+ */
public MetricsService(MetricsSystem metricsSystem, Executor executor) {
this.metricsSystem = metricsSystem;
try {
@@ -44,12 +50,14 @@ public class MetricsService {
server.setExecutor(executor);
}
+ /** Starts the HTTP server and exposes the Prometheus metrics endpoint. */
public void start() {
log.info("Starting Metrics Service for Prometheus ...");
server.createContext("/prometheus", metricsSystem.getPrometheusPullModelHandler());
server.start();
}
+ /** Stops the HTTP server. */
public void stop() {
log.info("Metrics Service stopped");
server.stop(0);
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/MetricsSystem.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/MetricsSystem.java
index 1b212dc..abd1ac1 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/MetricsSystem.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/MetricsSystem.java
@@ -48,10 +48,16 @@ public class MetricsSystem {
@Getter private final PrometheusPullModelHandler prometheusPullModelHandler;
private final Map<String, SinkProperties> sinkPropertiesMap;
+ /** Constructs a new MetricsSystem with default properties. */
public MetricsSystem() {
this(new Properties());
}
+ /**
+ * Constructs a new MetricsSystem with the given properties.
+ *
+ * @param properties The properties to configure the metrics system.
+ */
public MetricsSystem(Properties properties) {
this.sources = new HashSet<>();
this.sinks = new HashSet<>();
@@ -63,6 +69,10 @@ public class MetricsSystem {
this.sinks.add(prometheusPullModelHandler);
}
+ /**
+ * Starts the metrics system, registering default sources and configured sinks. Throws
+ * IllegalStateException if the system is already running.
+ */
public void start() {
if (running.get()) {
throw new IllegalStateException(
@@ -74,6 +84,7 @@ public class MetricsSystem {
sinks.forEach(Sink::start);
}
+ /** Stops the metrics system, stopping all sinks and clearing the registry. */
public void stop() {
if (running.get()) {
sinks.forEach(Sink::stop);
@@ -84,14 +95,20 @@ public class MetricsSystem {
running.set(false);
}
+ /** Triggers all registered sinks to report their metrics. */
public void report() {
sinks.forEach(Sink::report);
}
+ /** Registers default metric sources, such as JVM metrics. */
protected void registerDefaultSources() {
registerSource(new OperatorJvmSource());
}
+ /**
+ * Registers all configured metrics sinks based on the provided properties. Sinks are instantiated
+ * via reflection.
+ */
protected void registerSinks() {
log.info("sinkPropertiesMap: {}", sinkPropertiesMap);
sinkPropertiesMap
@@ -124,6 +141,11 @@ public class MetricsSystem {
});
}
+ /**
+ * Registers a custom metric source with the metrics system.
+ *
+ * @param source The Source to register.
+ */
public void registerSource(Source source) {
sources.add(source);
try {
@@ -139,6 +161,7 @@ public class MetricsSystem {
String className;
Properties properties;
+ /** Constructs a new, empty SinkProperties object. */
public SinkProperties() {
this.className = "";
this.properties = new Properties();
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/MetricsSystemFactory.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/MetricsSystemFactory.java
index 0eb1de9..15370e2 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/MetricsSystemFactory.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/MetricsSystemFactory.java
@@ -36,12 +36,23 @@ public final class MetricsSystemFactory {
private MetricsSystemFactory() {}
+ /**
+ * Creates a MetricsSystem instance using properties from SparkOperatorConfManager.
+ *
+ * @return A new MetricsSystem instance.
+ */
public static MetricsSystem createMetricsSystem() {
Properties properties =
parseMetricsProperties(SparkOperatorConfManager.INSTANCE.getMetricsProperties());
return new MetricsSystem(properties);
}
+ /**
+ * Creates a MetricsSystem instance with the given properties.
+ *
+ * @param properties The properties to configure the metrics system.
+ * @return A new MetricsSystem instance.
+ */
public static MetricsSystem createMetricsSystem(Properties properties) {
return new MetricsSystem(properties);
}
@@ -58,6 +69,12 @@ public final class MetricsSystemFactory {
return properties;
}
+ /**
+ * Parses the given properties to extract sink-related configurations.
+ *
+ * @param metricsProperties The properties containing metrics configurations.
+ * @return A Map where keys are sink short names and values are SinkProperties objects.
+ */
public static Map<String, MetricsSystem.SinkProperties> parseSinkProperties(
Properties metricsProperties) {
Map<String, MetricsSystem.SinkProperties> propertiesMap = new HashMap<>();
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/PrometheusPullModelHandler.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/PrometheusPullModelHandler.java
index 8592e18..736a845 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/PrometheusPullModelHandler.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/PrometheusPullModelHandler.java
@@ -54,6 +54,12 @@ public class PrometheusPullModelHandler extends PrometheusServlet implements Htt
@Getter private final boolean enablePrometheusTextBasedFormat;
@Getter private final boolean enableSanitizePrometheusMetricsName;
+ /**
+ * Constructs a new PrometheusPullModelHandler.
+ *
+ * @param properties Configuration properties.
+ * @param registry The MetricRegistry to pull metrics from.
+ */
public PrometheusPullModelHandler(Properties properties, MetricRegistry registry) {
super(properties, registry);
this.registry = registry;
@@ -63,16 +69,24 @@ public class PrometheusPullModelHandler extends PrometheusServlet implements Htt
SparkOperatorConf.SANITIZE_PROMETHEUS_METRICS_NAME_ENABLED.getValue();
}
+ /** Starts the Prometheus pull model handler. */
@Override
public void start() {
log.info("PrometheusPullModelHandler started");
}
+ /** Stops the Prometheus pull model handler. */
@Override
public void stop() {
log.info("PrometheusPullModelHandler stopped");
}
+ /**
+ * Handles HTTP requests for Prometheus metrics.
+ *
+ * @param exchange The HttpExchange object.
+ * @throws IOException if an I/O error occurs.
+ */
@Override
public void handle(HttpExchange exchange) throws IOException {
if (enablePrometheusTextBasedFormat) {
@@ -92,6 +106,12 @@ public class PrometheusPullModelHandler extends PrometheusServlet implements Htt
}
}
+ /**
+ * Filters out empty metric records from a Prometheus metrics snapshot string.
+ *
+ * @param metricsSnapshot The raw metrics snapshot string.
+ * @return A List of strings containing only non-empty metric records.
+ */
protected List<String> filterNonEmptyRecords(String metricsSnapshot) {
// filter out empty records which could cause Prometheus invalid syntax exception, e.g:
// metrics_jvm_threadStates_deadlocks_Number{type="gauges"} []
@@ -108,6 +128,11 @@ public class PrometheusPullModelHandler extends PrometheusServlet implements Htt
return filteredRecords;
}
+ /**
+ * Formats the current metrics snapshot into a Prometheus text-based format.
+ *
+ * @return A String containing the formatted metrics.
+ */
protected String formatMetricsSnapshot() {
Map<String, Gauge> gauges = registry.getGauges();
Map<String, Counter> counters = registry.getCounters();
@@ -145,12 +170,25 @@ public class PrometheusPullModelHandler extends PrometheusServlet implements Htt
return stringBuilder.toString();
}
+ /**
+ * Appends a string value to a StringBuilder if the value is not empty.
+ *
+ * @param stringBuilder The StringBuilder to append to.
+ * @param value The string value to append.
+ */
protected void appendIfNotEmpty(StringBuilder stringBuilder, String value) {
if (StringUtils.isNotEmpty(value)) {
stringBuilder.append(value);
}
}
+ /**
+ * Formats a Gauge metric into a Prometheus-compatible string.
+ *
+ * @param name The name of the gauge.
+ * @param gauge The Gauge object.
+ * @return A formatted string for the gauge, or null if the gauge is invalid.
+ */
protected String formatGauge(String name, Gauge gauge) {
if (gauge != null
&& gauge.getValue() != null
@@ -171,6 +209,13 @@ public class PrometheusPullModelHandler extends PrometheusServlet implements Htt
return null;
}
+ /**
+ * Formats a Counter metric into a Prometheus-compatible string.
+ *
+ * @param name The name of the counter.
+ * @param counter The Counter object.
+ * @return A formatted string for the counter, or null if the counter is invalid.
+ */
protected String formatCounter(String name, Counter counter) {
if (counter != null) {
String formattedName = sanitize(name);
@@ -188,6 +233,13 @@ public class PrometheusPullModelHandler extends PrometheusServlet implements Htt
return null;
}
+ /**
+ * Formats a Histogram metric into a Prometheus-compatible string.
+ *
+ * @param name The name of the histogram.
+ * @param histogram The Histogram object.
+ * @return A formatted string for the histogram, or null if the histogram is invalid.
+ */
protected String formatHistogram(String name, Histogram histogram) {
if (histogram != null && histogram.getSnapshot() != null) {
String baseName = sanitize(name);
@@ -242,6 +294,13 @@ public class PrometheusPullModelHandler extends PrometheusServlet implements Htt
return null;
}
+ /**
+ * Formats a Meter metric into a Prometheus-compatible string.
+ *
+ * @param name The name of the meter.
+ * @param meter The Meter object.
+ * @return A formatted string for the meter, or null if the meter is invalid.
+ */
protected String formatMeter(String name, Meter meter) {
if (meter != null) {
String baseName = sanitize(name);
@@ -272,6 +331,13 @@ public class PrometheusPullModelHandler extends PrometheusServlet implements Htt
return null;
}
+ /**
+ * Formats a Timer metric into a Prometheus-compatible string.
+ *
+ * @param name The name of the timer.
+ * @param timer The Timer object.
+ * @return A formatted string for the timer, or null if the timer is invalid.
+ */
protected String formatTimer(String name, Timer timer) {
if (timer != null && timer.getSnapshot() != null) {
String baseName = sanitize(name);
@@ -343,10 +409,22 @@ public class PrometheusPullModelHandler extends PrometheusServlet implements Htt
return null;
}
+ /**
+ * Converts nanoseconds to seconds.
+ *
+ * @param nanos The value in nanoseconds.
+ * @return The value in seconds.
+ */
protected double nanosToSeconds(double nanos) {
return nanos / 1_000_000_000.0;
}
+ /**
+ * Sanitizes a metric name to be Prometheus-compatible.
+ *
+ * @param name The original metric name.
+ * @return The sanitized metric name.
+ */
protected String sanitize(String name) {
if (enableSanitizePrometheusMetricsName) {
return name.replaceAll("[^a-zA-Z0-9_]", "_").toLowerCase();
@@ -354,6 +432,12 @@ public class PrometheusPullModelHandler extends PrometheusServlet implements Htt
return name;
}
+ /**
+ * Converts a metric name containing "_nanos" to "_seconds".
+ *
+ * @param name The original metric name.
+ * @return The converted metric name.
+ */
protected String nanosMetricsNameToSeconds(String name) {
return name.replaceAll("_nanos", "_seconds");
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/SparkAppStatusRecorderSource.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/SparkAppStatusRecorderSource.java
index 622b425..644ffaf 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/SparkAppStatusRecorderSource.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/SparkAppStatusRecorderSource.java
@@ -28,25 +28,43 @@ import org.apache.spark.k8s.operator.status.ApplicationState;
import org.apache.spark.k8s.operator.status.ApplicationStatus;
import org.apache.spark.metrics.source.Source;
+/** Metric source for recording Spark application status updates. */
public class SparkAppStatusRecorderSource extends BaseOperatorSource implements Source {
public static final String RESOURCE_TYPE = "sparkapp";
public static final String LATENCY_METRIC_FORMAT = "latency.from.%s.to.%s";
+ /** Constructs a new SparkAppStatusRecorderSource. */
public SparkAppStatusRecorderSource() {
super(new MetricRegistry());
}
+ /**
+ * Returns the name of the metric source.
+ *
+ * @return The name of the source.
+ */
@Override
public String sourceName() {
return "SparkAppStatusRecorder";
}
+ /**
+ * Returns the MetricRegistry associated with this source.
+ *
+ * @return The MetricRegistry.
+ */
@Override
public MetricRegistry metricRegistry() {
return metricRegistry;
}
+ /**
+ * Records the latency of a status update.
+ *
+ * @param status The current application status.
+ * @param newState The new application state.
+ */
public void recordStatusUpdateLatency(
final ApplicationStatus status, final ApplicationState newState) {
ApplicationState currentState = status.getCurrentState();
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/healthcheck/SentinelManager.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/healthcheck/SentinelManager.java
index ec5e32e..88fac43 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/healthcheck/SentinelManager.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/healthcheck/SentinelManager.java
@@ -58,6 +58,12 @@ public class SentinelManager<CR extends BaseResource<?, ?, ?, ?, ?>> {
Executors.newScheduledThreadPool(
SparkOperatorConf.SENTINEL_EXECUTOR_SERVICE_POOL_SIZE.getValue());
+ /**
+ * Checks if a given resource is a sentinel resource.
+ *
+ * @param resource The resource to check.
+ * @return True if the resource is a sentinel resource, false otherwise.
+ */
public static boolean isSentinelResource(HasMetadata resource) {
Map<String, String> labels = resource.getMetadata().getLabels();
if (labels == null) {
@@ -82,6 +88,11 @@ public class SentinelManager<CR extends BaseResource<?, ?, ?, ?, ?>> {
return true;
}
+ /**
+ * Checks if all tracked sentinel resources are healthy.
+ *
+ * @return True if all sentinels are healthy, false otherwise.
+ */
public boolean allSentinelsAreHealthy() {
Set<ResourceID> unWatchedKey = new HashSet<>();
boolean result =
@@ -101,6 +112,12 @@ public class SentinelManager<CR extends BaseResource<?, ?, ?, ?, ?>> {
return result;
}
+ /**
+ * Checks the health of a specific sentinel resource.
+ *
+ * @param resourceID The ID of the resource to check.
+ * @param client The Kubernetes client.
+ */
public void checkHealth(ResourceID resourceID, KubernetesClient client) {
SentinelResourceState sentinelResourceState = sentinelResources.get(resourceID);
if (sentinelResourceState == null) {
@@ -128,6 +145,13 @@ public class SentinelManager<CR extends BaseResource<?, ?, ?, ?, ?>> {
updateSpecAndScheduleHealthCheck(resourceID, sentinelResourceState, client);
}
+ /**
+ * Handles the reconciliation of a sentinel resource.
+ *
+ * @param resource The custom resource being reconciled.
+ * @param client The Kubernetes client.
+ * @return True if the resource was a sentinel resource and was handled, false otherwise.
+ */
public boolean handleSentinelResourceReconciliation(CR resource, KubernetesClient client) {
if (!isSentinelResource(resource)) {
return false;
@@ -191,14 +215,29 @@ public class SentinelManager<CR extends BaseResource<?, ?, ?, ?, ?>> {
@Getter boolean isHealthy = true;
+ /**
+ * Updates the internal resource with the latest custom resource.
+ *
+ * @param cr The latest custom resource.
+ */
void onReconcile(CR cr) {
resource = cr;
}
+ /**
+ * Checks if the resource has been reconciled since its last update.
+ *
+ * @return True if reconciled, false otherwise.
+ */
boolean reconciledSinceUpdate() {
return resource.getMetadata().getGeneration() > previousGeneration;
}
+ /**
+ * Returns a string representation of the SentinelResourceState.
+ *
+ * @return A string representation.
+ */
@Override
public String toString() {
return new ToStringBuilder(this)
@@ -209,6 +248,11 @@ public class SentinelManager<CR extends BaseResource<?, ?, ?, ?, ?>> {
}
}
+ /**
+ * Returns the map of sentinel resources for testing purposes.
+ *
+ * @return A ConcurrentHashMap of ResourceID to SentinelResourceState.
+ */
@VisibleForTesting
public ConcurrentHashMap<ResourceID, SentinelResourceState> getSentinelResources() {
return sentinelResources;
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java
index aab1c0e..685a6d9 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/KubernetesMetricsInterceptor.java
@@ -62,6 +62,7 @@ public class KubernetesMetricsInterceptor implements Interceptor, Source {
private final Meter responseRateMeter;
private final Map<String, Meter> namespacedResourceMethodMeters = new ConcurrentHashMap<>();
+ /** Constructs a new KubernetesMetricsInterceptor, initializing metric registries. */
public KubernetesMetricsInterceptor() {
metricRegistry = new MetricRegistry();
@@ -88,6 +89,13 @@ public class KubernetesMetricsInterceptor implements Interceptor, Source {
}
}
+ /**
+ * Intercepts an HTTP request and updates Kubernetes client metrics.
+ *
+ * @param chain The Interceptor.Chain for the current request.
+ * @return The Response from the intercepted request.
+ * @throws IOException if an I/O error occurs during the request.
+ */
@NotNull
@Override
public Response intercept(@NotNull Chain chain) throws IOException {
@@ -103,11 +111,21 @@ public class KubernetesMetricsInterceptor implements Interceptor, Source {
}
}
+ /**
+ * Returns the name of this metrics source.
+ *
+ * @return The source name.
+ */
@Override
public String sourceName() {
return "kubernetes.client";
}
+ /**
+ * Returns the MetricRegistry used by this interceptor.
+ *
+ * @return The MetricRegistry instance.
+ */
@Override
public MetricRegistry metricRegistry() {
return this.metricRegistry;
@@ -160,6 +178,12 @@ public class KubernetesMetricsInterceptor implements Interceptor, Source {
metricRegistry.meter(MetricRegistry.name(HTTP_RESPONSE_GROUP, String.valueOf(code))));
}
+ /**
+ * Parses the given path to extract namespace-scoped resource information.
+ *
+ * @param path The request path.
+ * @return An Optional containing a Pair of namespace and resource name, or empty if not found.
+ */
public Optional<Pair<String, String>> parseNamespaceScopedResource(String path) {
if (path.contains(NAMESPACES)) {
int index = path.indexOf(NAMESPACES) + NAMESPACES.length();
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/OperatorJosdkMetrics.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/OperatorJosdkMetrics.java
index 967a2d6..627716b 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/OperatorJosdkMetrics.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/OperatorJosdkMetrics.java
@@ -68,27 +68,49 @@ public class OperatorJosdkMetrics extends BaseOperatorSource implements Source,
private final Clock clock;
+ /** Constructs a new OperatorJosdkMetrics instance. */
public OperatorJosdkMetrics() {
super(new MetricRegistry());
this.clock = new SystemClock();
}
+ /**
+ * Returns the name of this metrics source.
+ *
+ * @return The source name.
+ */
@Override
public String sourceName() {
return PREFIX;
}
+ /**
+ * Returns the MetricRegistry used by this metrics source.
+ *
+ * @return The MetricRegistry instance.
+ */
@Override
public MetricRegistry metricRegistry() {
return metricRegistry;
}
+ /**
+ * Called when a controller is registered.
+ *
+ * @param controller The registered controller.
+ */
@Override
public void controllerRegistered(Controller<? extends HasMetadata> controller) {
// no-op
log.debug("Controller has been registered");
}
+ /**
+ * Records metrics when an event is received.
+ *
+ * @param event The received event.
+ * @param metadata Additional metadata associated with the event.
+ */
@Override
public void receivedEvent(Event event, Map<String, Object> metadata) {
log.debug("received event {}, metadata {}", event, metadata);
@@ -113,6 +135,14 @@ public class OperatorJosdkMetrics extends BaseOperatorSource implements Source,
}
}
+ /**
+ * Times the execution of a controller operation and records success or failure metrics.
+ *
+ * @param execution The ControllerExecution to time.
+ * @param <T> The return type of the execution.
+ * @return The result of the execution.
+ * @throws Exception if the execution throws an exception.
+ */
@Override
public <T> T timeControllerExecution(ControllerExecution<T> execution) throws Exception {
log.debug("Time controller execution");
@@ -166,6 +196,13 @@ public class OperatorJosdkMetrics extends BaseOperatorSource implements Source,
}
}
+ /**
+ * Records metrics related to custom resource reconciliation, including retry information.
+ *
+ * @param resource The custom resource being reconciled.
+ * @param retryInfo Information about retries, if any.
+ * @param metadata Additional metadata.
+ */
@Override
public void reconcileCustomResource(
HasMetadata resource, RetryInfo retryInfo, Map<String, Object> metadata) {
@@ -184,6 +221,13 @@ public class OperatorJosdkMetrics extends BaseOperatorSource implements Source,
.inc();
}
+ /**
+ * Records metrics when a reconciliation fails.
+ *
+ * @param resource The custom resource for which reconciliation failed.
+ * @param exception The exception that caused the failure.
+ * @param metadata Additional metadata.
+ */
@Override
public void failedReconciliation(
HasMetadata resource, Exception exception, Map<String, Object> metadata) {
@@ -194,6 +238,12 @@ public class OperatorJosdkMetrics extends BaseOperatorSource implements Source,
getCounter(metricsPrefix, resource.getMetadata().getNamespace(), RECONCILIATION, FAILED).inc();
}
+ /**
+ * Records metrics when a reconciliation finishes successfully.
+ *
+ * @param resource The custom resource for which reconciliation finished.
+ * @param metadata Additional metadata.
+ */
@Override
public void finishedReconciliation(HasMetadata resource, Map<String, Object> metadata) {
log.debug("Finished reconciliation for resource {} with metadata {}", resource, metadata);
@@ -202,6 +252,12 @@ public class OperatorJosdkMetrics extends BaseOperatorSource implements Source,
getCounter(metricsPrefix, resource.getMetadata().getNamespace(), RECONCILIATION, FINISHED);
}
+ /**
+ * Records metrics when cleanup is done for a resource.
+ *
+ * @param resourceID The ID of the resource that was cleaned up.
+ * @param metadata Additional metadata.
+ */
@Override
public void cleanupDoneFor(ResourceID resourceID, Map<String, Object> metadata) {
log.debug("Cleanup Done for resource {} with metadata {}", resourceID, metadata);
@@ -212,6 +268,14 @@ public class OperatorJosdkMetrics extends BaseOperatorSource implements Source,
.ifPresent(ns -> getCounter(metricsPrefix, ns, RECONCILIATION, CLEANUP).inc());
}
+ /**
+ * Monitors the size of a map and exposes it as a Gauge metric.
+ *
+ * @param map The map to monitor.
+ * @param name The name of the metric.
+ * @param <T> The type of the map.
+ * @return The monitored map.
+ */
@Override
public <T extends Map<?, ?>> T monitorSizeOf(T map, String name) {
log.debug("Monitor size for {}", name);
@@ -226,6 +290,12 @@ public class OperatorJosdkMetrics extends BaseOperatorSource implements Source,
return map;
}
+ /**
+ * Records metrics when a reconciliation execution starts.
+ *
+ * @param resource The custom resource for which reconciliation started.
+ * @param metadata Additional metadata.
+ */
@Override
public void reconciliationExecutionStarted(HasMetadata resource, Map<String, Object> metadata) {
log.debug("Reconciliation execution started");
@@ -241,6 +311,12 @@ public class OperatorJosdkMetrics extends BaseOperatorSource implements Source,
.inc();
}
+ /**
+ * Records metrics when a reconciliation execution finishes.
+ *
+ * @param resource The custom resource for which reconciliation finished.
+ * @param metadata Additional metadata.
+ */
@Override
public void reconciliationExecutionFinished(HasMetadata resource, Map<String, Object> metadata) {
log.debug("Reconciliation execution finished");
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/OperatorJvmSource.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/OperatorJvmSource.java
index ccbb478..1b6c27a 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/OperatorJvmSource.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/source/OperatorJvmSource.java
@@ -26,11 +26,21 @@ import org.apache.spark.metrics.source.Source;
/** Source for JVM metrics. */
public class OperatorJvmSource implements Source {
+ /**
+ * Returns the name of this metrics source.
+ *
+ * @return The source name.
+ */
@Override
public String sourceName() {
return "jvm";
}
+ /**
+ * Returns the MetricRegistry containing JVM metrics.
+ *
+ * @return The MetricRegistry instance.
+ */
@Override
public MetricRegistry metricRegistry() {
MetricRegistry metricRegistry = new MetricRegistry();
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/HealthProbe.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/HealthProbe.java
index 5b33a9e..750827a 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/HealthProbe.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/HealthProbe.java
@@ -51,6 +51,12 @@ public class HealthProbe implements HttpHandler {
private final List<Operator> operators;
private final List<SentinelManager<?>> sentinelManagers;
+ /**
+ * Checks the overall health of the operator, including all registered operators and sentinel
+ * managers.
+ *
+ * @return True if the operator is healthy, false otherwise.
+ */
public boolean isHealthy() {
Optional<Boolean> operatorsAreReady = areOperatorsStarted(operators);
if (operatorsAreReady.isEmpty() || !operatorsAreReady.get()) {
@@ -76,6 +82,12 @@ public class HealthProbe implements HttpHandler {
return true;
}
+ /**
+ * Handles HTTP requests for the health probe endpoint.
+ *
+ * @param exchange The HttpExchange object.
+ * @throws IOException if an I/O error occurs.
+ */
@Override
public void handle(HttpExchange exchange) throws IOException {
if (isHealthy()) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ProbeService.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ProbeService.java
index fa153c6..9bcd6bb 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ProbeService.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ProbeService.java
@@ -40,6 +40,13 @@ public class ProbeService {
public static final String READYZ = "/readyz";
@Getter private final HttpServer server;
+ /**
+ * Constructs a new ProbeService.
+ *
+ * @param operators A list of Operator instances to monitor.
+ * @param sentinelManagers A list of SentinelManager instances to monitor.
+ * @param executor The Executor to use for the HTTP server.
+ */
public ProbeService(
List<Operator> operators, List<SentinelManager<?>> sentinelManagers, Executor executor) {
try {
@@ -52,11 +59,13 @@ public class ProbeService {
server.setExecutor(executor);
}
+ /** Starts the probe service HTTP server. */
public void start() {
log.info("Probe service started at " + OPERATOR_PROBE_PORT.getValue());
server.start();
}
+ /** Stops the probe service HTTP server. */
public void stop() {
log.info("Probe service stopped");
server.stop(0);
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ReadinessProbe.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ReadinessProbe.java
index d66d9bf..a4778ca 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ReadinessProbe.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ReadinessProbe.java
@@ -37,10 +37,21 @@ import lombok.extern.slf4j.Slf4j;
public class ReadinessProbe implements HttpHandler {
private final List<Operator> operators;
+ /**
+ * Constructs a new ReadinessProbe.
+ *
+ * @param operators A list of Operator instances to check for readiness.
+ */
public ReadinessProbe(List<Operator> operators) {
this.operators = operators;
}
+ /**
+ * Handles HTTP requests for the readiness probe endpoint.
+ *
+ * @param httpExchange The HttpExchange object.
+ * @throws IOException if an I/O error occurs.
+ */
@Override
public void handle(HttpExchange httpExchange) throws IOException {
Optional<Boolean> operatorsAreReady = areOperatorsStarted(operators);
@@ -56,6 +67,11 @@ public class ReadinessProbe implements HttpHandler {
sendMessage(httpExchange, HTTP_OK, "started");
}
+ /**
+ * Checks if RBAC requirements are met. Currently always returns true.
+ *
+ * @return True if RBAC check passes, false otherwise.
+ */
public boolean passRbacCheck() {
return true;
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/ReconcileProgress.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/ReconcileProgress.java
index fa0879a..9047545 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/ReconcileProgress.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/ReconcileProgress.java
@@ -45,24 +45,55 @@ public final class ReconcileProgress {
this.requeueAfterDuration = requeueAfterDuration;
}
+ /**
+ * Creates a ReconcileProgress indicating that reconciliation should proceed and be re-queued
+ * after the default interval.
+ *
+ * @return A ReconcileProgress instance.
+ */
public static ReconcileProgress proceed() {
return new ReconcileProgress(
false, true, Duration.ofSeconds(RECONCILER_INTERVAL_SECONDS.getValue()));
}
+ /**
+ * Creates a ReconcileProgress indicating that reconciliation is complete and should be re-queued
+ * after the default interval.
+ *
+ * @return A ReconcileProgress instance.
+ */
public static ReconcileProgress completeAndDefaultRequeue() {
return new ReconcileProgress(
true, true, Duration.ofSeconds(RECONCILER_INTERVAL_SECONDS.getValue()));
}
+ /**
+ * Creates a ReconcileProgress indicating that reconciliation is complete and should be re-queued
+ * after a specified duration.
+ *
+ * @param requeueAfterDuration The duration after which to re-queue the reconciliation.
+ * @return A ReconcileProgress instance.
+ */
public static ReconcileProgress completeAndRequeueAfter(Duration requeueAfterDuration) {
return new ReconcileProgress(true, true, requeueAfterDuration);
}
+ /**
+ * Creates a ReconcileProgress indicating that reconciliation is complete and should be re-queued
+ * immediately.
+ *
+ * @return A ReconcileProgress instance.
+ */
public static ReconcileProgress completeAndImmediateRequeue() {
return new ReconcileProgress(true, true, Duration.ZERO);
}
+ /**
+ * Creates a ReconcileProgress indicating that reconciliation is complete and should not be
+ * re-queued.
+ *
+ * @return A ReconcileProgress instance.
+ */
public static ReconcileProgress completeAndNoRequeue() {
return new ReconcileProgress(true, false, Duration.ZERO);
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
index bae0e5b..e5ab729 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
@@ -75,6 +75,14 @@ public class SparkAppReconciler implements Reconciler<SparkApplication>, Cleaner
private final SparkAppStatusRecorder sparkAppStatusRecorder;
private final SentinelManager<SparkApplication> sentinelManager;
+ /**
+ * Reconciles the state of a SparkApplication resource.
+ *
+ * @param sparkApplication The SparkApplication resource to reconcile.
+ * @param context The reconciliation context.
+ * @return An UpdateControl object indicating the result of the reconciliation.
+ * @throws Exception if an error occurs during reconciliation.
+ */
@Override
public UpdateControl<SparkApplication> reconcile(
SparkApplication sparkApplication, Context<SparkApplication> context) throws Exception {
@@ -103,6 +111,14 @@ public class SparkAppReconciler implements Reconciler<SparkApplication>, Cleaner
}
}
+ /**
+ * Updates the error status of a SparkApplication resource.
+ *
+ * @param sparkApplication The SparkApplication resource.
+ * @param context The reconciliation context.
+ * @param e The exception that occurred.
+ * @return An ErrorStatusUpdateControl indicating no status update.
+ */
@Override
public ErrorStatusUpdateControl<SparkApplication> updateErrorStatus(
SparkApplication sparkApplication, Context<SparkApplication> context, Exception e) {
@@ -126,6 +142,12 @@ public class SparkAppReconciler implements Reconciler<SparkApplication>, Cleaner
}
}
+ /**
+ * Prepares the event sources for the SparkApplication reconciler.
+ *
+ * @param context The EventSourceContext.
+ * @return A List of EventSource objects.
+ */
@Override
public List<EventSource<?, SparkApplication>> prepareEventSources(
EventSourceContext<SparkApplication> context) {
@@ -140,6 +162,12 @@ public class SparkAppReconciler implements Reconciler<SparkApplication>, Cleaner
return List.of(podEventSource);
}
+ /**
+ * Returns a list of reconciliation steps based on the current state of the SparkApplication.
+ *
+ * @param app The SparkApplication resource.
+ * @return A List of AppReconcileStep objects.
+ */
protected List<AppReconcileStep> getReconcileSteps(final SparkApplication app) {
List<AppReconcileStep> steps = new ArrayList<>();
steps.add(new AppValidateStep());
@@ -171,11 +199,11 @@ public class SparkAppReconciler implements Reconciler<SparkApplication>, Cleaner
}
/**
- * Best-effort graceful termination upon delete.
+ * Performs cleanup operations for a SparkApplication resource when it is marked for deletion.
*
- * @param sparkApplication the resource that is marked for deletion
- * @param context the context with which the operation is executed
- * @return DeleteControl, with requeue if needed
+ * @param sparkApplication The SparkApplication resource that is marked for deletion.
+ * @param context The context with which the operation is executed.
+ * @return A DeleteControl object, with requeue if needed.
*/
@Override
public DeleteControl cleanup(
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppResourceSpecFactory.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppResourceSpecFactory.java
index 98ba0fb..a1c3353 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppResourceSpecFactory.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppResourceSpecFactory.java
@@ -50,6 +50,14 @@ public final class SparkAppResourceSpecFactory {
private SparkAppResourceSpecFactory() {}
+ /**
+ * Builds a SparkAppResourceSpec for the given SparkApplication.
+ *
+ * @param app The SparkApplication.
+ * @param client The KubernetesClient.
+ * @param worker The SparkAppSubmissionWorker.
+ * @return A SparkAppResourceSpec containing the configured resources.
+ */
public static SparkAppResourceSpec buildResourceSpec(
final SparkApplication app,
final KubernetesClient client,
@@ -62,6 +70,12 @@ public final class SparkAppResourceSpecFactory {
return resourceSpec;
}
+ /**
+ * Overrides dependency configurations based on the SparkApplication's spec.
+ *
+ * @param app The SparkApplication.
+ * @return A Map of configuration overrides.
+ */
private static Map<String, String> overrideDependencyConf(final SparkApplication app) {
Map<String, String> confOverrides = new HashMap<>();
sparkAppResourceLabels(app)
@@ -80,6 +94,12 @@ public final class SparkAppResourceSpecFactory {
return confOverrides;
}
+ /**
+ * Cleans up temporary resources created for the SparkApplication.
+ *
+ * @param app The SparkApplication.
+ * @param confOverrides The configuration overrides map.
+ */
private static void cleanUpTempResourcesForApp(
final SparkApplication app, Map<String, String> confOverrides) {
if (overrideDriverTemplateEnabled(app.getSpec())) {
@@ -90,6 +110,13 @@ public final class SparkAppResourceSpecFactory {
}
}
+ /**
+ * Retrieves a local file from a given path key in the configuration overrides.
+ *
+ * @param confOverrides The configuration overrides map.
+ * @param pathKey The key for the file path.
+ * @return An Optional containing the File object if found and valid, otherwise empty.
+ */
private static Optional<File> getLocalFileFromPathKey(
Map<String, String> confOverrides, String pathKey) {
if (confOverrides.containsKey(pathKey)) {
@@ -101,6 +128,12 @@ public final class SparkAppResourceSpecFactory {
return Optional.empty();
}
+ /**
+ * Deletes a local file specified by a path key in the configuration overrides.
+ *
+ * @param confOverrides The configuration overrides map.
+ * @param pathKey The key for the file path.
+ */
private static void deleteLocalFileFromPathKey(
Map<String, String> confOverrides, String pathKey) {
Optional<File> localFile = Optional.empty();
@@ -121,6 +154,14 @@ public final class SparkAppResourceSpecFactory {
}
}
+ /**
+ * Gets or creates a local file for the driver pod template specification.
+ *
+ * @param app The SparkApplication.
+ * @param confOverrides The configuration overrides map.
+ * @return A Map containing the path to the local file if created or found, otherwise an empty
+ * map.
+ */
private static Map<String, String> getOrCreateLocalFileForDriverSpec(
final SparkApplication app, final Map<String, String> confOverrides) {
if (overrideDriverTemplateEnabled(app.getSpec())) {
@@ -137,6 +178,14 @@ public final class SparkAppResourceSpecFactory {
return Collections.emptyMap();
}
+ /**
+ * Gets or creates a local file for the executor pod template specification.
+ *
+ * @param app The SparkApplication.
+ * @param confOverrides The configuration overrides map.
+ * @return A Map containing the path to the local file if created or found, otherwise an empty
+ * map.
+ */
private static Map<String, String> getOrCreateLocalFileForExecutorSpec(
final SparkApplication app, final Map<String, String> confOverrides) {
if (overrideExecutorTemplateEnabled(app.getSpec())) {
@@ -154,9 +203,11 @@ public final class SparkAppResourceSpecFactory {
}
/**
- * Flush driver pod template spec to a local file
+ * Flushes a PodTemplateSpec to a local temporary file.
*
- * @return temp file path
+ * @param podTemplateSpec The PodTemplateSpec to write to a file.
+ * @param tempFilePrefix The prefix for the temporary file name.
+ * @return The absolute path to the created temporary file.
*/
private static String createLocalFileForPodTemplateSpec(
final PodTemplateSpec podTemplateSpec, final String tempFilePrefix) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
index f99d558..a94d80a 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
@@ -63,6 +63,14 @@ public class SparkClusterReconciler implements Reconciler<SparkCluster>, Cleaner
private final SparkClusterStatusRecorder sparkClusterStatusRecorder;
private final SentinelManager<SparkCluster> sentinelManager;
+ /**
+ * Reconciles the state of a SparkCluster resource.
+ *
+ * @param sparkCluster The SparkCluster resource to reconcile.
+ * @param context The reconciliation context.
+ * @return An UpdateControl object indicating the result of the reconciliation.
+ * @throws Exception if an error occurs during reconciliation.
+ */
@Override
public UpdateControl<SparkCluster> reconcile(
SparkCluster sparkCluster, Context<SparkCluster> context) throws Exception {
@@ -88,6 +96,14 @@ public class SparkClusterReconciler implements Reconciler<SparkCluster>, Cleaner
}
}
+ /**
+ * Updates the error status of a SparkCluster resource.
+ *
+ * @param sparkCluster The SparkCluster resource.
+ * @param context The reconciliation context.
+ * @param e The exception that occurred.
+ * @return An ErrorStatusUpdateControl indicating no status update.
+ */
@Override
public ErrorStatusUpdateControl<SparkCluster> updateErrorStatus(
SparkCluster sparkCluster, Context<SparkCluster> context, Exception e) {
@@ -110,6 +126,12 @@ public class SparkClusterReconciler implements Reconciler<SparkCluster>, Cleaner
}
}
+ /**
+ * Prepares the event sources for the SparkCluster reconciler.
+ *
+ * @param context The EventSourceContext.
+ * @return A List of EventSource objects.
+ */
@Override
public List<EventSource<?, SparkCluster>> prepareEventSources(
EventSourceContext<SparkCluster> context) {
@@ -124,6 +146,12 @@ public class SparkClusterReconciler implements Reconciler<SparkCluster>, Cleaner
return List.of(podEventSource);
}
+ /**
+ * Returns a list of reconciliation steps based on the current state of the SparkCluster.
+ *
+ * @param cluster The SparkCluster resource.
+ * @return A List of ClusterReconcileStep objects.
+ */
protected List<ClusterReconcileStep> getReconcileSteps(final SparkCluster cluster) {
List<ClusterReconcileStep> steps = new ArrayList<>();
steps.add(new ClusterValidateStep());
@@ -139,11 +167,12 @@ public class SparkClusterReconciler implements Reconciler<SparkCluster>, Cleaner
}
/**
- * Best-effort graceful termination upon delete.
+ * Performs cleanup operations for a SparkCluster resource when it is marked for deletion. This is
+ * a Best-effort graceful termination upon delete.
*
- * @param sparkCluster the resource that is marked for deletion
- * @param context the context with which the operation is executed
- * @return DeleteControl, with requeue if needed
+ * @param sparkCluster The SparkCluster resource that is marked for deletion.
+ * @param context The context with which the operation is executed.
+ * @return A DeleteControl object, with requeue if needed.
*/
@Override
public DeleteControl cleanup(SparkCluster sparkCluster, Context<SparkCluster> context) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterResourceSpecFactory.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterResourceSpecFactory.java
index 45d6372..b63ecef 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterResourceSpecFactory.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterResourceSpecFactory.java
@@ -35,6 +35,13 @@ public final class SparkClusterResourceSpecFactory {
private SparkClusterResourceSpecFactory() {}
+ /**
+ * Builds a SparkClusterResourceSpec for the given SparkCluster.
+ *
+ * @param cluster The SparkCluster.
+ * @param worker The SparkClusterSubmissionWorker.
+ * @return A SparkClusterResourceSpec containing the configured resources.
+ */
public static SparkClusterResourceSpec buildResourceSpec(
final SparkCluster cluster, final SparkClusterSubmissionWorker worker) {
Map<String, String> confOverrides = new HashMap<>();
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverReadyObserver.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverReadyObserver.java
index 7a506f1..d587dec 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverReadyObserver.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverReadyObserver.java
@@ -33,6 +33,15 @@ import org.apache.spark.k8s.operator.utils.PodUtils;
/** Observes whether driver is ready. */
public class AppDriverReadyObserver extends BaseAppDriverObserver {
+ /**
+ * Observes the driver pod to determine if it is ready.
+ *
+ * @param driver The driver Pod object.
+ * @param spec The ApplicationSpec of the Spark application.
+ * @param status The current ApplicationStatus of the Spark application.
+ * @return An Optional containing the new ApplicationState if the driver is ready or terminated,
+ * otherwise empty.
+ */
@Override
public Optional<ApplicationState> observe(
Pod driver, ApplicationSpec spec, ApplicationStatus status) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverRunningObserver.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverRunningObserver.java
index ff6a408..c5a779e 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverRunningObserver.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverRunningObserver.java
@@ -33,6 +33,15 @@ import org.apache.spark.k8s.operator.status.ApplicationStatus;
* step.
*/
public class AppDriverRunningObserver extends BaseAppDriverObserver {
+ /**
+ * Observes the driver pod to determine if it is running.
+ *
+ * @param driver The driver Pod object.
+ * @param spec The ApplicationSpec of the Spark application.
+ * @param currentStatus The current ApplicationStatus of the Spark application.
+ * @return An Optional containing the new ApplicationState if the driver is terminated, otherwise
+ * empty.
+ */
@Override
public Optional<ApplicationState> observe(
Pod driver, ApplicationSpec spec, ApplicationStatus currentStatus) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverStartObserver.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverStartObserver.java
index 66c9257..21542a7 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverStartObserver.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverStartObserver.java
@@ -33,6 +33,15 @@ import org.apache.spark.k8s.operator.utils.PodUtils;
/** Observes whether the driver pod has started. */
public class AppDriverStartObserver extends BaseAppDriverObserver {
+ /**
+ * Observes the driver pod to determine if it has started.
+ *
+ * @param driver The driver Pod object.
+ * @param spec The ApplicationSpec of the Spark application.
+ * @param currentStatus The current ApplicationStatus of the Spark application.
+ * @return An Optional containing the new ApplicationState if the driver has started or
+ * terminated, otherwise empty.
+ */
@Override
public Optional<ApplicationState> observe(
Pod driver, ApplicationSpec spec, ApplicationStatus currentStatus) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverTimeoutObserver.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverTimeoutObserver.java
index 6b26c02..3633840 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverTimeoutObserver.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/AppDriverTimeoutObserver.java
@@ -35,8 +35,9 @@ import org.apache.spark.k8s.operator.utils.SparkAppStatusUtils;
public class AppDriverTimeoutObserver extends BaseAppDriverObserver {
/**
+ * Observes the driver status and applies timeouts as configured in the application specification.
* Operator may proactively terminate application if it has stay in certain state for a while.
- * This helps to avoid resource deadlock when app cannot proceed. Such states include:
+ * This helps to avoid resource deadlock when app cannot proceed. For example,
*
* <ul>
* <li>DriverRequested goes to DriverStartTimedOut if driver pod cannot be scheduled or cannot
@@ -50,6 +51,12 @@ public class AppDriverTimeoutObserver extends BaseAppDriverObserver {
* lose them. User may build additional layers to alert and act on such scenario. Timeout check
* would be performed at the end of reconcile - and it would be performed only if there's no other
* updates to be performed in the same reconcile action
+ *
+ * @param driver The driver Pod object.
+ * @param spec The ApplicationSpec of the Spark application.
+ * @param status The current ApplicationStatus of the Spark application.
+ * @return An Optional containing the new ApplicationState if a timeout has occurred, otherwise
+ * empty.
*/
@Override
public Optional<ApplicationState> observe(
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseAppDriverObserver.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseAppDriverObserver.java
index f8cc25c..5bb909c 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseAppDriverObserver.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseAppDriverObserver.java
@@ -70,11 +70,11 @@ public abstract class BaseAppDriverObserver
* <li>2. The container(s) has exited 0 after SparkContext / SparkSession initialization
* </ul>
*
- * @param driverPod the driverPod
- * @param driverReady whether SparkContext / SparkSession has ever been initialized for this pod
- * @param spec the application spec
- * @return the ApplicationState to be updated if pod is terminated. Returning empty if pod is
- * still running
+ * @param driverPod The driver Pod object.
+ * @param driverReady A boolean indicating if the SparkContext/SparkSession was initialized.
+ * @param spec The ApplicationSpec of the Spark application.
+ * @return An Optional containing the new ApplicationState if the driver is terminated, otherwise
+ * empty.
*/
protected Optional<ApplicationState> observeDriverTermination(
final Pod driverPod, final boolean driverReady, final ApplicationSpec spec) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseSecondaryResourceObserver.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseSecondaryResourceObserver.java
index 1bb3a96..6ae0e35 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseSecondaryResourceObserver.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseSecondaryResourceObserver.java
@@ -46,5 +46,13 @@ public abstract class BaseSecondaryResourceObserver<
SPEC extends BaseSpec,
STATUS extends BaseStatus<S, STATE, AS>,
SR extends HasMetadata> {
+ /**
+ * Observes a secondary resource and returns an updated state if applicable.
+ *
+ * @param secondaryResource The secondary resource to observe.
+ * @param spec The specification of the primary resource.
+ * @param currentStatus The current status of the primary resource.
+ * @return An Optional containing the updated state, or empty if no update is needed.
+ */
public abstract Optional<STATE> observe(SR secondaryResource, SPEC spec, STATUS currentStatus);
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java
index 12b241f..3d1291e 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java
@@ -59,6 +59,12 @@ public class AppCleanUpStep extends AppReconcileStep {
private Supplier<ApplicationState> onDemandCleanUpReason;
private String stateUpdateMessage;
+ /**
+ * Constructs an AppCleanUpStep with a specific reason for on-demand cleanup.
+ *
+ * @param onDemandCleanUpReason A Supplier that provides the ApplicationState for on-demand
+ * cleanup.
+ */
public AppCleanUpStep(Supplier<ApplicationState> onDemandCleanUpReason) {
super();
this.onDemandCleanUpReason = onDemandCleanUpReason;
@@ -83,9 +89,9 @@ public class AppCleanUpStep extends AppReconcileStep {
* when even the reconciler decides to proceed with clean up, sub-resources may still be retained
* based on tolerations.
*
- * @param context context for the app
- * @param statusRecorder recorder for status updates in this reconcile
- * @return the reconcile progress
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating the next step.
*/
@Override
public ReconcileProgress reconcile(
@@ -185,6 +191,13 @@ public class AppCleanUpStep extends AppReconcileStep {
}
}
+ /**
+ * Clears the status cache and indicates that reconciliation for the application is complete.
+ *
+ * @param application The SparkApplication to clear cache for.
+ * @param statusRecorder The SparkAppStatusRecorder.
+ * @return An Optional containing a ReconcileProgress to complete and not re-queue.
+ */
protected Optional<ReconcileProgress> clearCacheAndFinishReconcileForApplication(
final SparkApplication application, final SparkAppStatusRecorder statusRecorder) {
log.debug("Cleaning up status cache and stop reconciling for application.");
@@ -192,6 +205,16 @@ public class AppCleanUpStep extends AppReconcileStep {
return Optional.of(ReconcileProgress.completeAndNoRequeue());
}
+ /**
+ * Checks if an early exit from reconciliation is possible for a terminated application, based on
+ * resource retention policies and TTL settings.
+ *
+ * @param client The KubernetesClient.
+ * @param application The SparkApplication.
+ * @param statusRecorder The SparkAppStatusRecorder.
+ * @return An Optional containing a ReconcileProgress if an early exit is determined, otherwise
+ * empty.
+ */
protected Optional<ReconcileProgress> checkEarlyExitForTerminatedApp(
final KubernetesClient client,
final SparkApplication application,
@@ -254,14 +277,20 @@ public class AppCleanUpStep extends AppReconcileStep {
return Optional.empty();
}
+ /**
+ * Checks if an on-demand cleanup has been requested.
+ *
+ * @return True if on-demand cleanup is requested, false otherwise.
+ */
protected boolean isOnDemandCleanup() {
return onDemandCleanUpReason != null;
}
/**
- * @param status status of the application
- * @return The last observed state before termination if the app has terminated. If the app has
- * not terminated, return the last observed state
+ * Returns the last observed state of the application before it terminated.
+ *
+ * @param status The current ApplicationStatus.
+ * @return The ApplicationState before termination, or the current state if not terminated.
*/
protected ApplicationState getLastObservedStateBeforeTermination(final ApplicationStatus status) {
ApplicationState lastObservedState = status.getCurrentState();
@@ -274,6 +303,12 @@ public class AppCleanUpStep extends AppReconcileStep {
return lastObservedState;
}
+ /**
+ * Determines if resources should be released due to a scheduling failure attempt.
+ *
+ * @param status The current ApplicationStatus.
+ * @return True if resources should be released due to scheduling failure, false otherwise.
+ */
protected boolean isReleasingResourcesForSchedulingFailureAttempt(
final ApplicationStatus status) {
ApplicationState lastObservedState = getLastObservedStateBeforeTermination(status);
@@ -281,6 +316,14 @@ public class AppCleanUpStep extends AppReconcileStep {
lastObservedState.getCurrentStateSummary());
}
+ /**
+ * Determines whether to retain or release resources based on the resource retention policy and
+ * current application state.
+ *
+ * @param resourceRetainPolicy The ResourceRetainPolicy configured for the application.
+ * @param currentState The current ApplicationState.
+ * @return True if resources should be retained, false if they should be released.
+ */
protected boolean retainReleaseResourceForPolicyAndState(
ResourceRetainPolicy resourceRetainPolicy, ApplicationState currentState) {
return switch (resourceRetainPolicy) {
@@ -290,6 +333,12 @@ public class AppCleanUpStep extends AppReconcileStep {
};
}
+ /**
+ * Determines if force deletion should be enabled for the given SparkApplication.
+ *
+ * @param app The SparkApplication to check.
+ * @return True if force deletion is enabled, false otherwise.
+ */
protected boolean enableForceDelete(SparkApplication app) {
long timeoutThreshold =
app.getSpec()
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java
index 1e9f2cf..f8cbfce 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java
@@ -47,6 +47,13 @@ import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
/** Request all driver and its resources when starting an attempt. */
@Slf4j
public class AppInitStep extends AppReconcileStep {
+ /**
+ * Reconciles the application initialization step, creating driver and associated resources.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating the next step.
+ */
@Override
public ReconcileProgress reconcile(
SparkAppContext context, SparkAppStatusRecorder statusRecorder) {
@@ -122,6 +129,12 @@ public class AppInitStep extends AppReconcileStep {
}
}
+ /**
+ * Creates an ApplicationState indicating a scheduling failure due to resource creation failure.
+ *
+ * @param failedResource The resource that failed to be created.
+ * @return An ApplicationState object representing the creation failure.
+ */
private ApplicationState creationFailureState(HasMetadata failedResource) {
return new ApplicationState(
ApplicationStateSummary.SchedulingFailure,
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java
index 6c414d4..3502708 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java
@@ -39,9 +39,24 @@ import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
/** Basic reconcile step for application. */
public abstract class AppReconcileStep {
+ /**
+ * Reconciles a specific step for a Spark application.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating the next step.
+ */
public abstract ReconcileProgress reconcile(
SparkAppContext context, SparkAppStatusRecorder statusRecorder);
+ /**
+ * Observes the driver pod and updates the application status based on a list of observers.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status updates.
+ * @param observers A list of BaseAppDriverObserver instances to apply.
+ * @return The ReconcileProgress indicating the next step.
+ */
protected ReconcileProgress observeDriver(
final SparkAppContext context,
final SparkAppStatusRecorder statusRecorder,
@@ -72,6 +87,15 @@ public abstract class AppReconcileStep {
}
}
+ /**
+ * Updates the application status and re-queues the reconciliation after a specified duration.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status updates.
+ * @param updatedStatus The updated ApplicationStatus.
+ * @param requeueAfter The duration after which to re-queue.
+ * @return The ReconcileProgress indicating the re-queue.
+ */
protected ReconcileProgress updateStatusAndRequeueAfter(
SparkAppContext context,
SparkAppStatusRecorder statusRecorder,
@@ -81,6 +105,16 @@ public abstract class AppReconcileStep {
return ReconcileProgress.completeAndRequeueAfter(requeueAfter);
}
+ /**
+ * Appends a new state to the application status, persists it, and re-queues the reconciliation
+ * after a specified duration.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status updates.
+ * @param newState The new ApplicationState to append.
+ * @param requeueAfter The duration after which to re-queue.
+ * @return The ReconcileProgress indicating the re-queue.
+ */
protected ReconcileProgress appendStateAndRequeueAfter(
SparkAppContext context,
SparkAppStatusRecorder statusRecorder,
@@ -90,6 +124,15 @@ public abstract class AppReconcileStep {
return ReconcileProgress.completeAndRequeueAfter(requeueAfter);
}
+ /**
+ * Appends a new state to the application status, persists it, and immediately re-queues the
+ * reconciliation.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status updates.
+ * @param newState The new ApplicationState to append.
+ * @return The ReconcileProgress indicating an immediate re-queue.
+ */
protected ReconcileProgress appendStateAndImmediateRequeue(
SparkAppContext context, SparkAppStatusRecorder statusRecorder, ApplicationState newState) {
return appendStateAndRequeueAfter(context, statusRecorder, newState, Duration.ZERO);
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppResourceObserveStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppResourceObserveStep.java
index 83333f4..38c0ac5 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppResourceObserveStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppResourceObserveStep.java
@@ -34,6 +34,14 @@ public class AppResourceObserveStep extends AppReconcileStep {
private final List<BaseAppDriverObserver> observers;
+ /**
+ * Reconciles the application by observing secondary resources and updating the application
+ * status.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating the next step.
+ */
@Override
public ReconcileProgress reconcile(
final SparkAppContext context, final SparkAppStatusRecorder statusRecorder) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppRunningStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppRunningStep.java
index 524c1cf..55d3544 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppRunningStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppRunningStep.java
@@ -38,6 +38,13 @@ import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
/** Observe whether app acquires enough executors as configured in spec. */
public class AppRunningStep extends AppReconcileStep {
+ /**
+ * Reconciles the application's running state, checking if enough executors are acquired.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating the next step.
+ */
@Override
public ReconcileProgress reconcile(
SparkAppContext context, SparkAppStatusRecorder statusRecorder) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppUnknownStateStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppUnknownStateStep.java
index ef11374..09cf195 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppUnknownStateStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppUnknownStateStep.java
@@ -32,6 +32,13 @@ import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
/** Abnormal state handler. */
public class AppUnknownStateStep extends AppReconcileStep {
+ /**
+ * Reconciles the application when it is in an unknown state, marking it as failed.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating an immediate re-queue.
+ */
@Override
public ReconcileProgress reconcile(
SparkAppContext context, SparkAppStatusRecorder statusRecorder) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStep.java
index 86b62bb..f3f5e17 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppValidateStep.java
@@ -36,6 +36,13 @@ import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
/** Validates the submitted app. This can be re-factored into webhook in the future. */
@Slf4j
public class AppValidateStep extends AppReconcileStep {
+ /**
+ * Reconciles the application by validating its configuration and status.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating the next step.
+ */
@Override
public ReconcileProgress reconcile(
SparkAppContext context, SparkAppStatusRecorder statusRecorder) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java
index db2fc3d..2159b05 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java
@@ -43,6 +43,13 @@ import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
/** Request cluster master and its resources when starting an attempt. */
@Slf4j
public class ClusterInitStep extends ClusterReconcileStep {
+ /**
+ * Reconciles the cluster initialization step, creating master and worker resources.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @param statusRecorder The SparkClusterStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating the next step.
+ */
@Override
public ReconcileProgress reconcile(
SparkClusterContext context, SparkClusterStatusRecorder statusRecorder) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterReconcileStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterReconcileStep.java
index 0a3199c..c7265e6 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterReconcileStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterReconcileStep.java
@@ -29,9 +29,25 @@ import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
/** Basic reconcile step for cluster. */
public abstract class ClusterReconcileStep {
+ /**
+ * Reconciles a specific step for a Spark cluster.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @param statusRecorder The SparkClusterStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating the next step.
+ */
public abstract ReconcileProgress reconcile(
SparkClusterContext context, SparkClusterStatusRecorder statusRecorder);
+ /**
+ * Updates the cluster status and re-queues the reconciliation after a specified duration.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @param statusRecorder The SparkClusterStatusRecorder for recording status updates.
+ * @param updatedStatus The updated ClusterStatus.
+ * @param requeueAfter The duration after which to re-queue.
+ * @return The ReconcileProgress indicating the re-queue.
+ */
protected ReconcileProgress updateStatusAndRequeueAfter(
SparkClusterContext context,
SparkClusterStatusRecorder statusRecorder,
@@ -41,6 +57,16 @@ public abstract class ClusterReconcileStep {
return ReconcileProgress.completeAndRequeueAfter(requeueAfter);
}
+ /**
+ * Appends a new state to the cluster status, persists it, and re-queues the reconciliation after
+ * a specified duration.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @param statusRecorder The SparkClusterStatusRecorder for recording status updates.
+ * @param newState The new ClusterState to append.
+ * @param requeueAfter The duration after which to re-queue.
+ * @return The ReconcileProgress indicating the re-queue.
+ */
protected ReconcileProgress appendStateAndRequeueAfter(
SparkClusterContext context,
SparkClusterStatusRecorder statusRecorder,
@@ -50,6 +76,15 @@ public abstract class ClusterReconcileStep {
return ReconcileProgress.completeAndRequeueAfter(requeueAfter);
}
+ /**
+ * Appends a new state to the cluster status, persists it, and immediately re-queues the
+ * reconciliation.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @param statusRecorder The SparkClusterStatusRecorder for recording status updates.
+ * @param newState The new ClusterState to append.
+ * @return The ReconcileProgress indicating an immediate re-queue.
+ */
protected ReconcileProgress appendStateAndImmediateRequeue(
SparkClusterContext context,
SparkClusterStatusRecorder statusRecorder,
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterTerminatedStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterTerminatedStep.java
index 346a1df..5826c82 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterTerminatedStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterTerminatedStep.java
@@ -27,6 +27,14 @@ import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
/** Observes whether cluster is already terminated. If so, end the reconcile. */
public class ClusterTerminatedStep extends ClusterReconcileStep {
+ /**
+ * Reconciles the cluster when it is in a terminated state, ending the reconciliation process.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @param statusRecorder The SparkClusterStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating that reconciliation is complete and should not be
+ * re-queued.
+ */
@Override
public ReconcileProgress reconcile(
SparkClusterContext context, SparkClusterStatusRecorder statusRecorder) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterUnknownStateStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterUnknownStateStep.java
index 6e70cd8..591e796 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterUnknownStateStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterUnknownStateStep.java
@@ -28,6 +28,13 @@ import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
/** Abnormal state handler for clusters. */
public class ClusterUnknownStateStep extends ClusterReconcileStep {
+ /**
+ * Reconciles the cluster when it is in an unknown state, marking it as failed.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @param statusRecorder The SparkClusterStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating an immediate re-queue.
+ */
@Override
public ReconcileProgress reconcile(
SparkClusterContext context, SparkClusterStatusRecorder statusRecorder) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterValidateStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterValidateStep.java
index c3ed521..fc8596c 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterValidateStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterValidateStep.java
@@ -30,6 +30,13 @@ import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
/** Validates the submitted cluster. This can be re-factored into webhook in the future. */
@Slf4j
public class ClusterValidateStep extends ClusterReconcileStep {
+ /**
+ * Reconciles the cluster by validating its configuration and status.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @param statusRecorder The SparkClusterStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating the next step.
+ */
@Override
public ReconcileProgress reconcile(
SparkClusterContext context, SparkClusterStatusRecorder statusRecorder) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ClassLoadingUtils.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ClassLoadingUtils.java
index f6e98c0..69b6535 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ClassLoadingUtils.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ClassLoadingUtils.java
@@ -36,13 +36,13 @@ public final class ClassLoadingUtils {
private ClassLoadingUtils() {}
/**
- * load status listener classes for given type of BaseStatusListener
+ * Loads status listener classes for a given type of BaseStatusListener.
*
* @param clazz Class name of status listener. e.g. SparkAppStatusListener
* @param implementationClassNames Comma-separated implementation class names of given type of
* StatusListener
- * @return list of listeners loaded
* @param <T> Class of status listener. e.g. SparkAppStatusListener
+ * @return list of listeners loaded
*/
public static <T extends BaseStatusListener<?, ?>> List<T> getStatusListener(
Class<T> clazz, String implementationClassNames) {
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/LoggingUtils.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/LoggingUtils.java
index bbe30f0..d438c6f 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/LoggingUtils.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/LoggingUtils.java
@@ -30,11 +30,17 @@ import org.apache.spark.k8s.operator.status.ApplicationAttemptSummary;
/** Utility class for logging. */
public class LoggingUtils {
+ /** Utility class for managing MDC (Mapped Diagnostic Context) for logging. */
public static final class TrackedMDC {
public static final String AppAttemptIdKey = "resource.app.attemptId";
private final ReentrantLock lock = new ReentrantLock();
private final Set<String> keys = new HashSet<>();
+ /**
+ * Sets the MDC (Mapped Diagnostic Context) with the application attempt ID if available.
+ *
+ * @param application The SparkApplication object.
+ */
public void set(final SparkApplication application) {
if (application != null && application.getStatus() != null) {
try {
@@ -50,6 +56,7 @@ public class LoggingUtils {
}
}
+ /** Resets the MDC by removing all previously set keys. */
public void reset() {
try {
lock.lock();
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/PodPhase.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/PodPhase.java
index 74c7d66..151ef50 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/PodPhase.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/PodPhase.java
@@ -41,6 +41,12 @@ public enum PodPhase {
this.description = description;
}
+ /**
+ * Returns the PodPhase for a given Pod.
+ *
+ * @param pod The Pod object.
+ * @return The corresponding PodPhase, or UNKNOWN if the phase cannot be determined.
+ */
public static PodPhase getPhase(final Pod pod) {
if (pod == null || pod.getStatus() == null) {
return UNKNOWN;
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/PodUtils.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/PodUtils.java
index 81a4cf3..b71576a 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/PodUtils.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/PodUtils.java
@@ -35,7 +35,12 @@ public final class PodUtils {
private PodUtils() {}
- /** Determine whether given pod is up running and ready */
+ /**
+ * Determines whether the given pod is running and ready.
+ *
+ * @param pod The Pod object.
+ * @return True if the pod is running and ready, false otherwise.
+ */
public static boolean isPodReady(final Pod pod) {
if (!PodPhase.RUNNING.equals(PodPhase.getPhase(pod))) {
return false;
@@ -54,11 +59,12 @@ public final class PodUtils {
}
/**
- * Determine whether the driver pod is started. Driver is considered as 'started' if any of Spark
- * container is started and ready
+ * Determines whether the driver pod is started. Driver is considered as 'started' if any of Spark
+ * container is started and ready.
*
- * @param driver the driver pod
- * @param spec expected spec for the SparkApp
+ * @param driver The driver pod.
+ * @param spec The expected spec for the SparkApp.
+ * @return True if the driver pod is started, false otherwise.
*/
public static boolean isDriverPodStarted(final Pod driver, final ApplicationSpec spec) {
// Consider pod as 'started' if any of Spark container is started and ready
@@ -81,19 +87,34 @@ public final class PodUtils {
.anyMatch(ContainerStatus::getReady);
}
- /** Returns true if the given container has terminated */
+ /**
+ * Returns true if the given container has terminated.
+ *
+ * @param containerStatus The ContainerStatus object.
+ * @return True if the container has terminated, false otherwise.
+ */
public static boolean isContainerTerminated(final ContainerStatus containerStatus) {
return containerStatus != null
&& containerStatus.getState() != null
&& containerStatus.getState().getTerminated() != null;
}
- /** Returns true if the given container has ever restarted */
+ /**
+ * Returns true if the given container has ever restarted.
+ *
+ * @param containerStatus The ContainerStatus object.
+ * @return True if the container has restarted, false otherwise.
+ */
public static boolean isContainerRestarted(final ContainerStatus containerStatus) {
return containerStatus != null && containerStatus.getRestartCount() > 0;
}
- /** Returns true if the given container has exited with non-zero status */
+ /**
+ * Returns true if the given container has exited with a non-zero status.
+ *
+ * @param containerStatus The ContainerStatus object.
+ * @return True if the container has failed, false otherwise.
+ */
public static boolean isContainerFailed(final ContainerStatus containerStatus) {
return isContainerTerminated(containerStatus)
&& containerStatus.getState().getTerminated().getExitCode() > 0;
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ProbeUtil.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ProbeUtil.java
index 4211fc6..3677f42 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ProbeUtil.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ProbeUtil.java
@@ -76,6 +76,13 @@ public final class ProbeUtil {
}
}
+ /**
+ * Checks if all provided Operator instances have started.
+ *
+ * @param operators A List of Operator instances.
+ * @return An Optional containing True if all operators are started, False otherwise. Empty if the
+ * list is empty.
+ */
public static Optional<Boolean> areOperatorsStarted(List<Operator> operators) {
return operators.stream()
.map(
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java
index 3a901f0..368313d 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ReconcilerUtils.java
@@ -51,6 +51,16 @@ public final class ReconcilerUtils {
private ReconcilerUtils() {}
+ /**
+ * Converts a ReconcileProgress to an UpdateControl.
+ *
+ * @param resource The resource being reconciled.
+ * @param reconcileProgress The ReconcileProgress object.
+ * @param <S> The type of the status.
+ * @param <T> The type of the spec.
+ * @param <O> The type of the resource, extending BaseResource.
+ * @return An UpdateControl object.
+ */
public static <S, T, O extends BaseResource<?, ?, ?, ?, ?>> UpdateControl<O> toUpdateControl(
O resource, ReconcileProgress reconcileProgress) {
// reconciler already handled resource and status update, skip update at lower level
@@ -62,6 +72,16 @@ public final class ReconcilerUtils {
}
}
+ /**
+ * Converts a ReconcileProgress to a DeleteControl.
+ *
+ * @param resource The resource being reconciled.
+ * @param reconcileProgress The ReconcileProgress object.
+ * @param <S> The type of the status.
+ * @param <T> The type of the spec.
+ * @param <O> The type of the resource, extending BaseResource.
+ * @return A DeleteControl object.
+ */
public static <S, T, O extends BaseResource<?, ?, ?, ?, ?>> DeleteControl toDeleteControl(
O resource, ReconcileProgress reconcileProgress) {
if (reconcileProgress.isRequeue()) {
@@ -72,6 +92,14 @@ public final class ReconcilerUtils {
}
}
+ /**
+ * Gets or creates a secondary Kubernetes resource.
+ *
+ * @param client The KubernetesClient.
+ * @param resource The desired resource to get or create.
+ * @param <T> The type of the resource, extending HasMetadata.
+ * @return An Optional containing the created or existing resource.
+ */
public static <T extends HasMetadata> Optional<T> getOrCreateSecondaryResource(
final KubernetesClient client, final T resource) {
Optional<T> current = getResource(client, resource);
@@ -112,6 +140,13 @@ public final class ReconcilerUtils {
return current;
}
+ /**
+ * Adds an owner reference to a list of secondary resources, linking them to a primary owner.
+ *
+ * @param client The KubernetesClient.
+ * @param resources The List of HasMetadata resources to modify.
+ * @param owner The primary owner resource.
+ */
public static void addOwnerReferenceSecondaryResource(
final KubernetesClient client, final List<HasMetadata> resources, final HasMetadata owner) {
@@ -126,6 +161,14 @@ public final class ReconcilerUtils {
client.resourceList(resources).forceConflicts().serverSideApply();
}
+ /**
+ * Retrieves a Kubernetes resource by its desired state.
+ *
+ * @param client The KubernetesClient.
+ * @param desired The desired state of the resource.
+ * @param <T> The type of the resource, extending HasMetadata.
+ * @return An Optional containing the retrieved resource, or empty if not found.
+ */
public static <T extends HasMetadata> Optional<T> getResource(
final KubernetesClient client, final T desired) {
T resource = null;
@@ -139,6 +182,14 @@ public final class ReconcilerUtils {
return Optional.ofNullable(resource);
}
+ /**
+ * Deletes a Kubernetes resource if it exists.
+ *
+ * @param client The KubernetesClient.
+ * @param resource The resource to delete.
+ * @param forceDelete If true, force deletes the resource with a grace period of 0.
+ * @param <T> The type of the resource, extending HasMetadata.
+ */
public static <T extends HasMetadata> void deleteResourceIfExists(
final KubernetesClient client, final T resource, boolean forceDelete) {
try {
@@ -160,6 +211,13 @@ public final class ReconcilerUtils {
}
}
+ /**
+ * Clones an object using JSON serialization and deserialization.
+ *
+ * @param object The object to clone.
+ * @param <T> The type of the object.
+ * @return A deep copy of the object.
+ */
public static <T> T clone(T object) {
if (object == null) {
return null;
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkAppStatusRecorder.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkAppStatusRecorder.java
index e9f224e..2123aaf 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkAppStatusRecorder.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkAppStatusRecorder.java
@@ -33,12 +33,25 @@ public class SparkAppStatusRecorder
extends StatusRecorder<ApplicationStatus, SparkApplication, SparkAppStatusListener> {
protected final SparkAppStatusRecorderSource recorderSource;
+ /**
+ * Constructs a new SparkAppStatusRecorder.
+ *
+ * @param statusListeners A list of SparkAppStatusListener instances.
+ * @param recorderSource The SparkAppStatusRecorderSource for metrics.
+ */
public SparkAppStatusRecorder(
List<SparkAppStatusListener> statusListeners, SparkAppStatusRecorderSource recorderSource) {
super(statusListeners, ApplicationStatus.class, SparkApplication.class);
this.recorderSource = recorderSource;
}
+ /**
+ * Appends a new state to the application status, records latency, and persists the updated
+ * status.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param newState The new ApplicationState to append.
+ */
public void appendNewStateAndPersist(SparkAppContext context, ApplicationState newState) {
ApplicationStatus appStatus = context.getResource().getStatus();
recorderSource.recordStatusUpdateLatency(appStatus, newState);
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkAppStatusUtils.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkAppStatusUtils.java
index 8b763bf..576c704 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkAppStatusUtils.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkAppStatusUtils.java
@@ -29,6 +29,12 @@ public final class SparkAppStatusUtils {
private SparkAppStatusUtils() {}
+ /**
+ * Checks if the application status is valid (not null and has a current state).
+ *
+ * @param app The SparkApplication to check.
+ * @return True if the application status is valid, false otherwise.
+ */
public static boolean isValidApplicationStatus(SparkApplication app) {
// null check
return app.getStatus() != null
@@ -36,36 +42,73 @@ public final class SparkAppStatusUtils {
&& app.getStatus().getCurrentState().getCurrentStateSummary() != null;
}
+ /**
+ * Creates an ApplicationState indicating that the driver was unexpectedly removed.
+ *
+ * @return An ApplicationState object for driver unexpected removal.
+ */
public static ApplicationState driverUnexpectedRemoved() {
return new ApplicationState(
ApplicationStateSummary.Failed, Constants.DRIVER_UNEXPECTED_REMOVED_MESSAGE);
}
+ /**
+ * Creates an ApplicationState indicating that the driver launch timed out.
+ *
+ * @return An ApplicationState object for driver launch timeout.
+ */
public static ApplicationState driverLaunchTimedOut() {
return new ApplicationState(
ApplicationStateSummary.DriverStartTimedOut, Constants.DRIVER_LAUNCH_TIMEOUT_MESSAGE);
}
+ /**
+ * Creates an ApplicationState indicating that the driver ready state timed out.
+ *
+ * @return An ApplicationState object for driver ready timeout.
+ */
public static ApplicationState driverReadyTimedOut() {
return new ApplicationState(
ApplicationStateSummary.DriverReadyTimedOut, Constants.DRIVER_LAUNCH_TIMEOUT_MESSAGE);
}
+ /**
+ * Creates an ApplicationState indicating that executor launch timed out.
+ *
+ * @return An ApplicationState object for executor launch timeout.
+ */
public static ApplicationState executorLaunchTimedOut() {
return new ApplicationState(
ApplicationStateSummary.ExecutorsStartTimedOut, Constants.EXECUTOR_LAUNCH_TIMEOUT_MESSAGE);
}
+ /**
+ * Creates an ApplicationState indicating that the application was cancelled.
+ *
+ * @return An ApplicationState object for application cancellation.
+ */
public static ApplicationState appCancelled() {
return new ApplicationState(
ApplicationStateSummary.ResourceReleased, Constants.APP_CANCELLED_MESSAGE);
}
+ /**
+ * Creates an ApplicationState indicating that the application exceeded its retain duration.
+ *
+ * @return An ApplicationState object for exceeding retain duration.
+ */
public static ApplicationState appExceededRetainDuration() {
return new ApplicationState(
ApplicationStateSummary.ResourceReleased, Constants.APP_EXCEEDED_RETAIN_DURATION_MESSAGE);
}
+ /**
+ * Checks if the application has reached a specific state in its transition history.
+ *
+ * @param application The SparkApplication to check.
+ * @param stateToCheck The ApplicationState to look for.
+ * @return True if the application has reached the state, false otherwise.
+ */
public static boolean hasReachedState(
SparkApplication application, ApplicationState stateToCheck) {
return isValidApplicationStatus(application)
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkClusterStatusRecorder.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkClusterStatusRecorder.java
index 82808c3..fa49793 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkClusterStatusRecorder.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkClusterStatusRecorder.java
@@ -30,10 +30,21 @@ import org.apache.spark.k8s.operator.status.ClusterStatus;
/** Records the status of a Spark cluster. */
public class SparkClusterStatusRecorder
extends StatusRecorder<ClusterStatus, SparkCluster, SparkClusterStatusListener> {
+ /**
+ * Constructs a new SparkClusterStatusRecorder.
+ *
+ * @param statusListeners A list of SparkClusterStatusListener instances.
+ */
public SparkClusterStatusRecorder(List<SparkClusterStatusListener> statusListeners) {
super(statusListeners, ClusterStatus.class, SparkCluster.class);
}
+ /**
+ * Appends a new state to the cluster status and persists the updated status.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @param newState The new ClusterState to append.
+ */
public void appendNewStateAndPersist(SparkClusterContext context, ClusterState newState) {
ClusterStatus updatedStatus = context.getResource().getStatus().appendNewState(newState);
persistStatus(context, updatedStatus);
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkExceptionUtils.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkExceptionUtils.java
index dabac32..cf7b3c9 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkExceptionUtils.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkExceptionUtils.java
@@ -30,6 +30,12 @@ public final class SparkExceptionUtils {
private SparkExceptionUtils() {}
+ /**
+ * Checks if a KubernetesClientException is a conflict due to an already existing resource.
+ *
+ * @param e The KubernetesClientException to check.
+ * @return True if it's a conflict for an existing resource, false otherwise.
+ */
public static boolean isConflictForExistingResource(KubernetesClientException e) {
return e != null
&& e.getCode() == HTTP_CONFLICT
@@ -38,6 +44,12 @@ public final class SparkExceptionUtils {
&& e.getStatus().toString().toLowerCase().contains("alreadyexists");
}
+ /**
+ * Builds a general error message from an Exception, including its stack trace.
+ *
+ * @param e The Exception to build the message from.
+ * @return A string containing the stack trace of the exception.
+ */
public static String buildGeneralErrorMessage(Exception e) {
return ExceptionUtils.getStackTrace(e);
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/StatusRecorder.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/StatusRecorder.java
index 01d669a..f3f8f75 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/StatusRecorder.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/StatusRecorder.java
@@ -58,6 +58,13 @@ public class StatusRecorder<
protected final Class<CR> resourceClass;
protected final ConcurrentHashMap<ResourceID, ObjectNode> statusCache;
+ /**
+ * Constructs a new StatusRecorder.
+ *
+ * @param statusListeners A list of status listeners.
+ * @param statusClass The Class object for the status type.
+ * @param resourceClass The Class object for the custom resource type.
+ */
protected StatusRecorder(
List<LISTENER> statusListeners, Class<STATUS> statusClass, Class<CR> resourceClass) {
this.statusListeners = statusListeners;
@@ -72,7 +79,8 @@ public class StatusRecorder<
* underlying resource spec was update in the meantime. This is necessary for the correct operator
* behavior.
*
- * @param resource Resource for which status update should be performed
+ * @param resource Resource for which status update should be performed.
+ * @param client KubernetesClient instance.
*/
@SneakyThrows
private void patchAndStatusWithVersionLocked(CR resource, KubernetesClient client) {
@@ -114,6 +122,12 @@ public class StatusRecorder<
});
}
+ /**
+ * Persists the new status of the resource to the Kubernetes cluster.
+ *
+ * @param context The BaseContext containing the resource and client.
+ * @param newStatus The new status to persist.
+ */
public void persistStatus(BaseContext<CR> context, STATUS newStatus) {
context.getResource().setStatus(newStatus);
patchAndStatusWithVersionLocked(context.getResource(), context.getClient());
@@ -127,7 +141,7 @@ public class StatusRecorder<
* <p>If the cache doesn't have a status stored, we do no update. This happens when the operator
* reconciles a resource for the first time after a restart.
*
- * @param resource Resource for which the status should be updated from the cache
+ * @param resource Resource for which the status should be updated from the cache.
*/
public void updateStatusFromCache(CR resource) {
ResourceID key = ResourceID.fromResource(resource);
@@ -140,7 +154,11 @@ public class StatusRecorder<
}
}
- /** Remove cached status */
+ /**
+ * Removes the cached status for a given custom resource.
+ *
+ * @param resource The custom resource for which to remove the cached status.
+ */
public void removeCachedStatus(CR resource) {
statusCache.remove(ResourceID.fromResource(resource));
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/Utils.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/Utils.java
index d152352..a91497b 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/Utils.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/Utils.java
@@ -54,6 +54,13 @@ public final class Utils {
private Utils() {}
+ /**
+ * Sanitizes a comma-separated string into a Set of trimmed, non-blank strings. If the input is
+ * "*", an empty set is returned.
+ *
+ * @param str The comma-separated string.
+ * @return A Set of sanitized strings.
+ */
public static Set<String> sanitizeCommaSeparatedStrAsSet(String str) {
if (StringUtils.isBlank(str)) {
return Collections.emptySet();
@@ -67,52 +74,103 @@ public final class Utils {
.collect(Collectors.toSet());
}
+ /**
+ * Converts a Map of labels to a comma-separated string of key=value pairs.
+ *
+ * @param labels The Map of labels.
+ * @return A string representation of the labels.
+ */
public static String labelsAsStr(Map<String, String> labels) {
return labels.entrySet().stream()
.map(e -> String.join("=", e.getKey(), e.getValue()))
.collect(Collectors.joining(","));
}
+ /**
+ * Returns a Map of common labels for operator resources.
+ *
+ * @return A Map of common operator resource labels.
+ */
public static Map<String, String> commonOperatorResourceLabels() {
Map<String, String> labels = new HashMap<>();
labels.put(LABEL_RESOURCE_NAME, OPERATOR_APP_NAME.getValue());
return labels;
}
+ /**
+ * Returns a Map of default labels for operator configuration resources.
+ *
+ * @return A Map of default operator configuration labels.
+ */
public static Map<String, String> defaultOperatorConfigLabels() {
Map<String, String> labels = new HashMap<>(commonOperatorResourceLabels());
labels.put("app.kubernetes.io/component", "operator-dynamic-config-overrides");
return labels;
}
+ /**
+ * Returns a Map of common labels for resources managed by the operator.
+ *
+ * @return A Map of common managed resource labels.
+ */
public static Map<String, String> commonManagedResourceLabels() {
Map<String, String> labels = new HashMap<>();
labels.put(LABEL_SPARK_OPERATOR_NAME, OPERATOR_APP_NAME.getValue());
return labels;
}
+ /**
+ * Returns a Map of labels for a SparkApplication resource.
+ *
+ * @param app The SparkApplication object.
+ * @return A Map of labels for the SparkApplication.
+ */
public static Map<String, String> sparkAppResourceLabels(final SparkApplication app) {
return sparkAppResourceLabels(app.getMetadata().getName());
}
+ /**
+ * Returns a Map of labels for a SparkApplication resource given its name.
+ *
+ * @param appName The name of the SparkApplication.
+ * @return A Map of labels for the SparkApplication.
+ */
public static Map<String, String> sparkAppResourceLabels(final String appName) {
Map<String, String> labels = commonManagedResourceLabels();
labels.put(Constants.LABEL_SPARK_APPLICATION_NAME, appName);
return labels;
}
+ /**
+ * Returns a Map of labels for the driver pod of a SparkApplication.
+ *
+ * @param app The SparkApplication object.
+ * @return A Map of labels for the driver pod.
+ */
public static Map<String, String> driverLabels(final SparkApplication app) {
Map<String, String> labels = sparkAppResourceLabels(app);
labels.put(Constants.LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_DRIVER_VALUE);
return labels;
}
+ /**
+ * Returns a Map of labels for executor pods of a SparkApplication.
+ *
+ * @param app The SparkApplication object.
+ * @return A Map of labels for executor pods.
+ */
public static Map<String, String> executorLabels(final SparkApplication app) {
Map<String, String> labels = sparkAppResourceLabels(app);
labels.put(Constants.LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_EXECUTOR_VALUE);
return labels;
}
+ /**
+ * Returns a Map of labels for a SparkCluster resource.
+ *
+ * @param cluster The SparkCluster object.
+ * @return A Map of labels for the SparkCluster.
+ */
public static Map<String, String> sparkClusterResourceLabels(final SparkCluster cluster) {
Map<String, String> labels = commonManagedResourceLabels();
labels.put(Constants.LABEL_SPARK_CLUSTER_NAME, cluster.getMetadata().getName());
@@ -120,35 +178,64 @@ public final class Utils {
return labels;
}
+ /**
+ * Returns a Map of labels for a SparkCluster's components.
+ *
+ * @param cluster The SparkCluster object.
+ * @return A Map of labels for the SparkCluster's components.
+ */
public static Map<String, String> clusterLabels(final SparkCluster cluster) {
Map<String, String> labels = sparkClusterResourceLabels(cluster);
labels.put(Constants.LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_CLUSTER_VALUE);
return labels;
}
+ /**
+ * Returns the set of namespaces that the operator is configured to watch.
+ *
+ * @return A Set of namespace names.
+ */
public static Set<String> getWatchedNamespaces() {
return sanitizeCommaSeparatedStrAsSet(OPERATOR_WATCHED_NAMESPACES.getValue());
}
+ /**
+ * Retrieves a list of SparkAppStatusListener instances based on configuration.
+ *
+ * @return A List of SparkAppStatusListener objects.
+ */
public static List<SparkAppStatusListener> getAppStatusListener() {
return ClassLoadingUtils.getStatusListener(
SparkAppStatusListener.class, SPARK_APP_STATUS_LISTENER_CLASS_NAMES.getValue());
}
+ /**
+ * Retrieves a list of SparkClusterStatusListener instances based on configuration.
+ *
+ * @return A List of SparkClusterStatusListener objects.
+ */
public static List<SparkClusterStatusListener> getClusterStatusListener() {
return ClassLoadingUtils.getStatusListener(
SparkClusterStatusListener.class, SPARK_CLUSTER_STATUS_LISTENER_CLASS_NAMES.getValue());
}
/**
- * Labels to be applied to all created resources, as a comma-separated string
+ * Returns a comma-separated string of common labels to be applied to all created resources.
*
- * @return labels string
+ * @return A string of common resource labels.
*/
public static String commonResourceLabelsStr() {
return labelsAsStr(commonManagedResourceLabels());
}
+ /**
+ * Creates a SecondaryToPrimaryMapper that maps secondary resources to their primary resources
+ * based on a common label.
+ *
+ * @param nameKey The label key used to identify the primary resource's name.
+ * @param <T> The type of the secondary resource, extending HasMetadata.
+ * @return A SecondaryToPrimaryMapper instance.
+ */
public static <T extends HasMetadata>
SecondaryToPrimaryMapper<T> basicLabelSecondaryToPrimaryMapper(String nameKey) {
return resource -> {
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppDriverConf.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppDriverConf.java
index cbe3f43..15f3a02 100644
--- a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppDriverConf.java
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppDriverConf.java
@@ -40,6 +40,17 @@ public final class SparkAppDriverConf extends KubernetesDriverConf {
super(sparkConf, appId, mainAppResource, mainClass, appArgs, proxyUser, null);
}
+ /**
+ * Creates a new SparkAppDriverConf instance.
+ *
+ * @param sparkConf The SparkConf object.
+ * @param appId The application ID.
+ * @param mainAppResource The main application resource.
+ * @param mainClass The main class of the application.
+ * @param appArgs The application arguments.
+ * @param proxyUser The proxy user option.
+ * @return A new SparkAppDriverConf instance.
+ */
public static SparkAppDriverConf create(
SparkConf sparkConf,
String appId,
@@ -53,20 +64,24 @@ public final class SparkAppDriverConf extends KubernetesDriverConf {
return new SparkAppDriverConf(sparkConf, appId, mainAppResource, mainClass, appArgs, proxyUser);
}
- /** Application managed by operator has a deterministic prefix */
+ /**
+ * Returns the resource name prefix, which is the application ID.
+ *
+ * @return The application ID.
+ */
@Override
public String resourceNamePrefix() {
return appId();
}
/**
- * Create the name to be used by driver config map. The consists of `resourceNamePrefix` and Spark
- * instance type (driver). Operator proposes `resourceNamePrefix` with leaves naming length margin
- * for sub-resources to be qualified as DNS subdomain or label. In addition, the overall config
- * name length is governed by `KubernetesClientUtils.configMapName` - which ensures the name
- * length meets requirements as DNS subdomain name.
+ * Creates the name to be used by the driver config map. The name consists of `resourceNamePrefix`
+ * and Spark instance type (driver). Operator proposes `resourceNamePrefix` with leaves naming
+ * length margin for sub-resources to be qualified as DNS subdomain or label. In addition, the
+ * overall config name length is governed by `KubernetesClientUtils.configMapName` - which ensures
+ * the name length meets the requirements as DNS subdomain name.
*
- * @return proposed name to be used by driver config map
+ * @return The proposed name for the driver config map.
*/
public String configMapNameDriver() {
return KubernetesClientUtils.configMapName(String.format("%s-spark-drv", resourceNamePrefix()));
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java
index e21647d..8fb5be2 100644
--- a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java
@@ -65,6 +65,14 @@ public class SparkAppResourceSpec {
@Getter private final List<HasMetadata> driverResources;
private final SparkAppDriverConf kubernetesDriverConf;
+ /**
+ * Constructs a new SparkAppResourceSpec.
+ *
+ * @param kubernetesDriverConf The KubernetesDriverConf for the application.
+ * @param kubernetesDriverSpec The KubernetesDriverSpec for the application.
+ * @param driverServiceIngressList A list of DriverServiceIngressSpec for the driver service.
+ * @param configMapSpecs A list of ConfigMapSpec for additional config maps.
+ */
public SparkAppResourceSpec(
SparkAppDriverConf kubernetesDriverConf,
KubernetesDriverSpec kubernetesDriverSpec,
@@ -102,6 +110,12 @@ public class SparkAppResourceSpec {
this.driverResources.forEach(r -> setNamespaceIfMissing(r, namespace));
}
+ /**
+ * Sets the namespace for a given resource if it's not already set.
+ *
+ * @param resource The resource to set the namespace for.
+ * @param namespace The namespace to set.
+ */
private void setNamespaceIfMissing(HasMetadata resource, String namespace) {
if (StringUtils.isNotEmpty(resource.getMetadata().getNamespace())) {
return;
@@ -109,6 +123,13 @@ public class SparkAppResourceSpec {
resource.getMetadata().setNamespace(namespace);
}
+ /**
+ * Adds a ConfigMap volume and mount to the SparkPod.
+ *
+ * @param pod The SparkPod to modify.
+ * @param confFilesMap The map of configuration files.
+ * @return The modified SparkPod.
+ */
private SparkPod addConfigMap(SparkPod pod, Map<String, String> confFilesMap) {
Container containerWithConfigMapVolume =
new ContainerBuilder(pod.container())
@@ -139,6 +160,13 @@ public class SparkAppResourceSpec {
return new SparkPod(podWithConfigMapVolume, containerWithConfigMapVolume);
}
+ /**
+ * Configures driver service ingress resources.
+ *
+ * @param pod The SparkPod for the driver.
+ * @param driverServiceIngressList A list of DriverServiceIngressSpec.
+ * @return A List of HasMetadata objects representing the ingress resources.
+ */
private List<HasMetadata> configureDriverServerIngress(
SparkPod pod, List<DriverServiceIngressSpec> driverServiceIngressList) {
if (driverServiceIngressList == null || driverServiceIngressList.isEmpty()) {
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
index 1a3ebe2..285e44d 100644
--- a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
@@ -47,17 +47,25 @@ import org.apache.spark.k8s.operator.utils.ModelUtils;
* SparkApplication instead of starting separate spark-submit process.
*/
public class SparkAppSubmissionWorker {
- // Default length limit for generated app id. Generated id is used as resource-prefix when
- // user-provided id is too long for this purpose. This applied to all resources associated with
- // the Spark app (including k8s service which has different naming length limit). Thus, we
- // truncate the hash part to 46 chars to leave some margin for spark resource prefix and suffix
- // (e.g. 'spark-', '-driver-svc' . etc)
+ /**
+ * Default length limit for generated app ID. Generated ID is used as resource-prefix when
+ * user-provided ID is too long for this purpose. This applied to all resources associated with
+ * the Spark app (including k8s service which has different naming length limit). Thus, we
+ * truncate the hash part to 46 chars to leave some margin for spark resource prefix and suffix
+ * (e.g. 'spark-', '-driver-svc' . etc)
+ */
public static final int DEFAULT_ID_LENGTH_LIMIT = 46;
- // Default length limit to be applied to the hash-based part of generated id
+
+ /** Default length limit to be applied to the hash-based part of generated ID. */
public static final int DEFAULT_HASH_BASED_IDENTIFIER_LENGTH_LIMIT = 36;
- // Radix value used when generating hash-based identifier
+
+ /** Radix value used when generating hash-based identifier. */
public static final int DEFAULT_ENCODE_BASE = 36;
+
+ /** Default master URL prefix for Kubernetes. */
public static final String DEFAULT_MASTER_URL_PREFIX = "k8s://";
+
+ /** Property name for the Spark master URL prefix. */
public static final String MASTER_URL_PREFIX_PROPS_NAME = "spark.master.url.prefix";
/**
@@ -77,10 +85,10 @@ public class SparkAppSubmissionWorker {
* master url based on it.
* </ul>
*
- * @param app the SparkApp resource
- * @param client k8s client
- * @param confOverrides key-value pairs of conf overrides for the given app.
- * @return secondary resources spec as `SparkAppResourceSpec`
+ * @param app The SparkApplication resource.
+ * @param client The Kubernetes client.
+ * @param confOverrides Key-value pairs of configuration overrides for the application.
+ * @return SparkAppResourceSpec containing the secondary resources.
*/
public SparkAppResourceSpec getResourceSpec(
SparkApplication app, KubernetesClient client, Map<String, String> confOverrides) {
@@ -92,6 +100,13 @@ public class SparkAppSubmissionWorker {
client);
}
+ /**
+ * Builds the SparkAppDriverConf for the given SparkApplication.
+ *
+ * @param app The SparkApplication.
+ * @param confOverrides Configuration overrides for the application.
+ * @return A SparkAppDriverConf instance.
+ */
protected SparkAppDriverConf buildDriverConf(
SparkApplication app, Map<String, String> confOverrides) {
ApplicationSpec applicationSpec = app.getSpec();
@@ -133,6 +148,15 @@ public class SparkAppSubmissionWorker {
Option.apply(applicationSpec.getProxyUser()));
}
+ /**
+ * Builds the SparkAppResourceSpec from the driver configuration and other specifications.
+ *
+ * @param kubernetesDriverConf The SparkAppDriverConf.
+ * @param driverServiceIngressList A list of DriverServiceIngressSpec.
+ * @param configMapSpecs A list of ConfigMapSpec.
+ * @param client The KubernetesClient.
+ * @return A SparkAppResourceSpec instance.
+ */
protected SparkAppResourceSpec buildResourceSpec(
SparkAppDriverConf kubernetesDriverConf,
List<DriverServiceIngressSpec> driverServiceIngressList,
@@ -146,8 +170,11 @@ public class SparkAppSubmissionWorker {
}
/**
- * Generate a Spark application id. Similar to `KubernetesConf.getKubernetesAppId()`. This is
- * deterministic per attempt per Spark App in order to ensure operator reconciliation idempotency
+ * Generates a deterministic Spark application ID based on the application's metadata and attempt
+ * ID in order to ensure operator reconciliation idempotency.
+ *
+ * @param app The SparkApplication object.
+ * @return A generated Spark application ID.
*/
public static String generateSparkAppId(final SparkApplication app) {
long attemptId = ModelUtils.getAttemptId(app);
@@ -167,13 +194,12 @@ public class SparkAppSubmissionWorker {
}
/**
- * Generate a hash-based id with given identifiers. The hash part would have a length-limit of
- * `DEFAULT_HASH_BASED_IDENTIFIER_LENGTH_LIMIT`. In addition, prefix would be applied upon
- * generated hash.
+ * Generates a hash-based ID with a given prefix and identifiers. The hash part would have a
+ * length-limit of `DEFAULT_HASH_BASED_IDENTIFIER_LENGTH_LIMIT`.
*
- * @param prefix string prefix to be applied to generated hash-based id.
- * @param identifiers keys to generate hash
- * @return generated hash-based id
+ * @param prefix String prefix to be applied to the generated hash-based ID.
+ * @param identifiers Keys to generate the hash.
+ * @return The generated hash-based ID.
*/
public static String generateHashBasedId(final String prefix, final String... identifiers) {
String sha256Hash =
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterResourceSpec.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterResourceSpec.java
index 69337d0..1c98273 100644
--- a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterResourceSpec.java
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterResourceSpec.java
@@ -54,6 +54,12 @@ public class SparkClusterResourceSpec {
@Getter private final StatefulSet workerStatefulSet;
@Getter private final Optional<HorizontalPodAutoscaler> horizontalPodAutoscaler;
+ /**
+ * Constructs a new SparkClusterResourceSpec.
+ *
+ * @param cluster The SparkCluster object.
+ * @param conf The SparkConf object.
+ */
public SparkClusterResourceSpec(SparkCluster cluster, SparkConf conf) {
String clusterNamespace = cluster.getMetadata().getNamespace();
String clusterName = cluster.getMetadata().getName();
@@ -106,6 +112,16 @@ public class SparkClusterResourceSpec {
horizontalPodAutoscaler = buildHorizontalPodAutoscaler(clusterName, namespace, spec);
}
+ /**
+ * Builds a Kubernetes Service for the Spark master.
+ *
+ * @param name The name of the cluster.
+ * @param namespace The namespace of the cluster.
+ * @param version The Spark version.
+ * @param metadata The ObjectMeta for the service.
+ * @param serviceSpec The ServiceSpec for the service.
+ * @return A Service object for the master.
+ */
private static Service buildMasterService(
String name, String namespace, String version, ObjectMeta metadata, ServiceSpec serviceSpec) {
return new ServiceBuilder()
@@ -138,6 +154,16 @@ public class SparkClusterResourceSpec {
.build();
}
+ /**
+ * Builds a Kubernetes Service for the Spark workers.
+ *
+ * @param name The name of the cluster.
+ * @param namespace The namespace of the cluster.
+ * @param version The Spark version.
+ * @param metadata The ObjectMeta for the service.
+ * @param serviceSpec The ServiceSpec for the service.
+ * @return A Service object for the workers.
+ */
private static Service buildWorkerService(
String name, String namespace, String version, ObjectMeta metadata, ServiceSpec serviceSpec) {
return new ServiceBuilder()
@@ -160,6 +186,19 @@ public class SparkClusterResourceSpec {
.build();
}
+ /**
+ * Builds a Kubernetes StatefulSet for the Spark master.
+ *
+ * @param scheduler The scheduler name.
+ * @param name The name of the cluster.
+ * @param namespace The namespace of the cluster.
+ * @param version The Spark version.
+ * @param image The container image.
+ * @param options Spark options string.
+ * @param objectMeta The ObjectMeta for the StatefulSet.
+ * @param statefulSetSpec The StatefulSetSpec for the StatefulSet.
+ * @return A StatefulSet object for the master.
+ */
private static StatefulSet buildMasterStatefulSet(
String scheduler,
String name,
@@ -225,6 +264,20 @@ public class SparkClusterResourceSpec {
.build();
}
+ /**
+ * Builds a Kubernetes StatefulSet for the Spark workers.
+ *
+ * @param scheduler The scheduler name.
+ * @param name The name of the cluster.
+ * @param namespace The namespace of the cluster.
+ * @param version The Spark version.
+ * @param image The container image.
+ * @param initWorkers The initial number of workers.
+ * @param options Spark options string.
+ * @param metadata The ObjectMeta for the StatefulSet.
+ * @param statefulSetSpec The StatefulSetSpec for the StatefulSet.
+ * @return A StatefulSet object for the workers.
+ */
private static StatefulSet buildWorkerStatefulSet(
String scheduler,
String name,
@@ -293,6 +346,14 @@ public class SparkClusterResourceSpec {
.build();
}
+ /**
+ * Builds a Kubernetes HorizontalPodAutoscaler for the Spark workers.
+ *
+ * @param clusterName The name of the cluster.
+ * @param namespace The namespace of the cluster.
+ * @param spec The ClusterSpec for the cluster.
+ * @return An Optional containing a HorizontalPodAutoscaler object, or empty if HPA is disabled.
+ */
private static Optional<HorizontalPodAutoscaler> buildHorizontalPodAutoscaler(
String clusterName, String namespace, ClusterSpec spec) {
var instanceConfig = spec.getClusterTolerations().getInstanceConfig();
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterSubmissionWorker.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterSubmissionWorker.java
index 0193f62..2460d82 100644
--- a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterSubmissionWorker.java
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterSubmissionWorker.java
@@ -27,6 +27,13 @@ import org.apache.spark.SparkConf;
/** Worker for submitting Spark clusters. */
public class SparkClusterSubmissionWorker {
+ /**
+ * Builds a SparkClusterResourceSpec for the given SparkCluster.
+ *
+ * @param cluster The SparkCluster object.
+ * @param confOverrides Configuration overrides for the cluster.
+ * @return A SparkClusterResourceSpec instance.
+ */
public SparkClusterResourceSpec getResourceSpec(
SparkCluster cluster, Map<String, String> confOverrides) {
SparkConf effectiveSparkConf = new SparkConf();
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/ConfigMapSpecUtils.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/ConfigMapSpecUtils.java
index 8c01ef2..49496c2 100644
--- a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/ConfigMapSpecUtils.java
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/ConfigMapSpecUtils.java
@@ -32,6 +32,12 @@ import org.apache.spark.k8s.operator.spec.ConfigMapSpec;
public final class ConfigMapSpecUtils {
private ConfigMapSpecUtils() {}
+ /**
+ * Builds a list of Kubernetes ConfigMap objects from a list of ConfigMapSpec.
+ *
+ * @param configMapMountList A list of ConfigMapSpec objects.
+ * @return A List of ConfigMap objects.
+ */
public static List<ConfigMap> buildConfigMaps(List<ConfigMapSpec> configMapMountList) {
if (configMapMountList == null || configMapMountList.isEmpty()) {
return Collections.emptyList();
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/DriverServiceIngressUtils.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/DriverServiceIngressUtils.java
index b794b96..ca2f0da 100644
--- a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/DriverServiceIngressUtils.java
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/DriverServiceIngressUtils.java
@@ -39,7 +39,13 @@ import org.apache.spark.k8s.operator.spec.DriverServiceIngressSpec;
public final class DriverServiceIngressUtils {
private DriverServiceIngressUtils() {}
- /** Build the full spec for ingress and service. */
+ /**
+ * Builds the full specification for ingress and service resources for a driver.
+ *
+ * @param spec The DriverServiceIngressSpec.
+ * @param driverPodMetaData The ObjectMeta of the driver pod.
+ * @return A List of HasMetadata objects representing the ingress and service.
+ */
public static List<HasMetadata> buildIngressService(
DriverServiceIngressSpec spec, ObjectMeta driverPodMetaData) {
List<HasMetadata> resources = new ArrayList<>(2);
@@ -49,6 +55,14 @@ public final class DriverServiceIngressUtils {
return resources;
}
+ /**
+ * Builds a Kubernetes Service object based on the provided specifications.
+ *
+ * @param spec The DriverServiceIngressSpec.
+ * @param driverPodMetaData The ObjectMeta of the driver pod, used for namespace and default
+ * selectors.
+ * @return A Service object.
+ */
private static Service buildService(DriverServiceIngressSpec spec, ObjectMeta driverPodMetaData) {
ObjectMeta serviceMeta = new ObjectMetaBuilder(spec.getServiceMetadata()).build();
serviceMeta.setNamespace(driverPodMetaData.getNamespace());
@@ -64,6 +78,13 @@ public final class DriverServiceIngressUtils {
.build();
}
+ /**
+ * Builds a Kubernetes Ingress object based on the provided specifications and associated Service.
+ *
+ * @param spec The DriverServiceIngressSpec.
+ * @param service The Service object that the Ingress will route to.
+ * @return An Ingress object.
+ */
private static Ingress buildIngress(DriverServiceIngressSpec spec, Service service) {
ObjectMeta metadata = new ObjectMetaBuilder(spec.getIngressMetadata()).build();
IngressSpec ingressSpec = new IngressSpecBuilder(spec.getIngressSpec()).build();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org