feedback changes

Author: bubriks

java/hsfs/src/main/java/com/logicalclocks/hsfs/JobConfiguration.java

@@ -17,6 +17,9 @@
 
 package com.logicalclocks.hsfs;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import com.google.common.base.Strings;
 
 import lombok.AllArgsConstructor;
@@ -28,8 +31,12 @@
 @AllArgsConstructor
 public class JobConfiguration {
 
+  private static Map<String, String> DEFAULT_PROPERTIES = new HashMap<String, String>() {{
+      put("spark.yarn.maxAppAttempts", "2");
+  }};
+
+
   @Getter
-  @Setter
   private String type = "sparkJobConfiguration";
 
 
@@ -73,11 +80,14 @@ public class JobConfiguration {
   private String properties;
 
   public String getProperties() {
-    String defaultProperty = "spark.yarn.maxAppAttempts=2";
-    if (Strings.isNullOrEmpty(properties)) {
-      properties = defaultProperty;
-    } else {
-      properties = properties + (properties.contains(defaultProperty.split("=")[0]) ? "" : "\n" + defaultProperty);
+    // Add default properties to the properties
+    for (Map.Entry<String, String> entry : JobConfiguration.DEFAULT_PROPERTIES.entrySet()) {
+      String defaultProperty = entry.getKey() + "=" + entry.getValue();
+      if (Strings.isNullOrEmpty(properties)) {
+        properties = defaultProperty;
+      } else if (!properties.contains(entry.getKey())) {
+        properties = properties + "\n" + defaultProperty;
+      }
     }
 
     return properties;

java/hsfs/src/test/java/com/logicalclocks/hsfs/TestJobConfiguration.java

@@ -0,0 +1,62 @@
+/*
+ *  Copyright (c) 2025. Hopsworks AB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *
+ *  See the License for the specific language governing permissions and limitations under the License.
+ *
+ */
+
+package com.logicalclocks.hsfs;
+
+import org.junit.Assert;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+
+class TestJobConfiguration {
+  @Test
+  void testGetProperties() throws FeatureStoreException, IOException {
+    // Arrange
+    JobConfiguration jobConfiguration = new JobConfiguration(null, 1, 2, 3, 4, 0, false, 0, 0, 0, null);
+
+    // Act
+    String result = jobConfiguration.getProperties();
+
+    // Assert
+    Assert.assertEquals("spark.yarn.maxAppAttempts=2", result);
+  }
+
+  @Test
+  void testGetPropertiesProvidedProperties() throws FeatureStoreException, IOException {
+    // Arrange
+    JobConfiguration jobConfiguration = new JobConfiguration(null, 1, 2, 3, 4, 0, false, 0, 0, 0, "spark.test=xxx");
+
+    // Act
+    String result = jobConfiguration.getProperties();
+
+    // Assert
+    Assert.assertEquals("spark.test=xxx\nspark.yarn.maxAppAttempts=2", result);
+  }
+
+  @Test
+  void testGetPropertiesProvidedPropertiesOverlapDefault() throws FeatureStoreException, IOException {
+    // Arrange
+    JobConfiguration jobConfiguration = new JobConfiguration(null, 1, 2, 3, 4, 0, false, 0, 0, 0, "spark.yarn.maxAppAttempts=9");
+
+    // Act
+    String result = jobConfiguration.getProperties();
+
+    // Assert
+    Assert.assertEquals("spark.yarn.maxAppAttempts=9", result);
+  }
+}

python/hsfs/core/ingestion_job_conf.py

@@ -28,7 +28,8 @@ def __init__(
         self._data_format = data_format
         self._data_options = data_options
         self._write_options = write_options
-        self._spark_job_configuration = spark_job_configuration
+
+        self._spark_job_configuration = JobConfiguration(**spark_job_configuration if spark_job_configuration else {}).to_dict(),
 
     @property
     def data_format(self):
@@ -74,5 +75,5 @@ def to_dict(self):
             ]
             if self._write_options
             else None,
-            JobConfiguration.DTO_TYPE: JobConfiguration(**self._spark_job_configuration if self._spark_job_configuration else {}).to_dict(),
+            JobConfiguration.DTO_TYPE: self._spark_job_configuration,
         }

python/hsfs/core/job_configuration.py

@@ -22,6 +22,7 @@
 
 class JobConfiguration:
     DTO_TYPE = "sparkJobConfiguration"
+    DEFAULT_PROPERTIES = {"spark.yarn.maxAppAttempts": 2}
 
     def __init__(
         self,
@@ -33,7 +34,7 @@ def __init__(
         dynamic_allocation=True,
         dynamic_min_executors=1,
         dynamic_max_executors=2,
-        properties=None,
+        properties="",
         **kwargs,
     ):
         self._am_memory = am_memory
@@ -44,15 +45,17 @@ def __init__(
         self._dynamic_allocation = dynamic_allocation
         self._dynamic_min_executors = dynamic_min_executors
         self._dynamic_max_executors = dynamic_max_executors
-        self._properties = properties
 
-    def to_dict(self):
-        default_property = "spark.yarn.maxAppAttempts=2"
-        if not self._properties:
-            self._properties = default_property
+        # Add default properties to the properties
+        default_properties = "\n".join([
+            f"{key}={value}" if not properties or key not in properties else ""
+            for key, value in JobConfiguration.DEFAULT_PROPERTIES.items()])
+        if properties:
+            self._properties = properties + ("\n" + default_properties if default_properties else "")
         else:
-            self._properties = self._properties + (f"\n{default_property}" if default_property.split("=")[0] not in self._properties else "")
+            self._properties = default_properties
 
+    def to_dict(self):
         return {
             "amMemory": self._am_memory,
             "amCores": self._am_cores,

python/hsfs/core/training_dataset_job_conf.py

@@ -26,7 +26,8 @@ def __init__(self, query, overwrite, write_options, spark_job_configuration):
         self._query = query
         self._overwrite = overwrite
         self._write_options = write_options
-        self._spark_job_configuration = spark_job_configuration
+
+        self._spark_job_configuration = JobConfiguration(**spark_job_configuration if spark_job_configuration else {}).to_dict(),
 
     @property
     def query(self):
@@ -72,5 +73,5 @@ def to_dict(self):
             ]
             if self._write_options
             else None,
-            JobConfiguration.DTO_TYPE: JobConfiguration(**self._spark_job_configuration if self._spark_job_configuration else {}).to_dict(),
+            JobConfiguration.DTO_TYPE: self._spark_job_configuration,
         }