Improve YAML parsing robustness and handle null Avro types (#202)

* Wrap Dynamics.newFromYaml() to handle malformed YAML

snakeyaml throws IndexOutOfBoundsException when parsing malformed YAML
via Dynamics.newFromYaml(). The call in estimateElementStatus() was
outside the existing try/catch, causing the exception to propagate up
through the JDBC layer and crash getPipelineStatus requests.

- Wrap Dynamics.newFromYaml() in estimateElementStatus() with try/catch;
  return a default unready status on parse failure instead of propagating
- Same fix in getElementConfiguration() which had the identical pattern
- Add tests for both: malformed YAML returns unready status / empty map

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: Handle SnakeYAML parsing exceptions in operator

SnakeYAML throws ScannerException, ParserException, and ConstructorException
when parsing malformed YAML in Kubernetes resource specs or annotations. Two
call sites were unprotected:

1. Operator.isReady(String yaml) and Operator.isFailed(String yaml): the
   Dynamics.newFromYaml() call was outside the existing try/catch, so any
   parse error would propagate up and crash the SubscriptionReconciler
   (stack trace through SubscriptionReconciler.reconcile line 171 ->
   Operator.isReady line 180). 

2. SubscriptionReconciler.fetchAttributes(String yaml): same unprotected
   Dynamics.newFromYaml() call, returns empty map on parse failure instead.

The K8sPipelineElementStatusEstimator call site was already fixed in a prior commit; this commit adds additional tests covering ScannerException,
ParserException, and ConstructorException for completeness.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

* fix: Guard against NullPointerException in K8sPipelineElementStatusEstimator

When Dynamics.newFromYaml() parses YAML without a metadata field, the
resulting DynamicKubernetesObject has null metadata. Calling
obj.getMetadata().getName() on it caused NPE in estimateElementStatus().

Guard against null metadata by returning an unready status early,
consistent with how we handle other parse/retrieval failures.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

* fix: handle null Avro schemas in AvroConverter to prevent NPE

Add null guard at entry to AvroConverter.rel() and explicit NULL type
handling. When a null schema or NULL-typed schema is encountered,
returns a proper SQL NULL type instead of throwing NPE.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: downgrade YAML exception logs ERROR→WARN, add null-safety guards

- K8sPipelineElementStatusEstimator: downgrade YAML parse errors from
  ERROR to WARN — exceptions are caught and handled correctly, ERROR
  level was causing spurious automated tickets
- Operator.isReady/isFailed: guard against null metadata and null
  namespace in K8s resource
- SubscriptionReconciler.fetchAttributes: guard against null metadata;
  downgrade SQL validation errors to WARN level

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Clean up comments & tests

* Build fixes - forgot to commit

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: jogrogan

hoptimator-avro/src/main/java/com/linkedin/hoptimator/avro/AvroConverter.java

@@ -163,8 +163,13 @@ public static RelDataType rel(Schema schema, RelDataTypeFactory typeFactory) {
    * TODO: default field values are lost when converting from Avro to RelDataType
    */
   public static RelDataType rel(Schema schema, RelDataTypeFactory typeFactory, boolean nullable) {
+    if (schema == null) {
+      return typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.NULL), true);
+    }
     RelDataType unknown = typeFactory.createUnknownType();
     switch (schema.getType()) {
+      case NULL:
+        return typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.NULL), true);
       case RECORD:
         return typeFactory.createTypeWithNullability(typeFactory.createStructType(schema.getFields().stream()
             .map(x -> new AbstractMap.SimpleEntry<>(x.name(), rel(x.schema(), typeFactory, nullable)))

hoptimator-avro/src/test/java/com/linkedin/hoptimator/avro/AvroConverterTest.java

@@ -217,6 +217,55 @@ public void convertsNestedArray() {
     assertEquals(Schema.Type.STRING, innermostField.getTypes().get(1).getType());
   }
 
+  @Test
+  public void convertsNullableUnionFields() {
+    // Schema with a nullable string field: ["null", "string"]
+    String schemaString =
+        "{\"type\":\"record\",\"name\":\"R\",\"namespace\":\"ns\",\"fields\":["
+            + "{\"name\":\"nullableStr\",\"type\":[\"null\",\"string\"]},"
+            + "{\"name\":\"nullableInt\",\"type\":[\"null\",\"int\"]},"
+            + "{\"name\":\"requiredStr\",\"type\":\"string\"}]}";
+    Schema avroSchema = (new Schema.Parser()).parse(schemaString);
+    RelDataType rel = AvroConverter.rel(avroSchema);
+    assertNotNull(rel);
+    assertEquals(3, rel.getFieldCount());
+    assertNotNull(rel.getField("nullableStr", false, false));
+    assertTrue(Objects.requireNonNull(rel.getField("nullableStr", false, false)).getType().isNullable());
+    assertNotNull(rel.getField("nullableInt", false, false));
+    assertTrue(Objects.requireNonNull(rel.getField("nullableInt", false, false)).getType().isNullable());
+    assertNotNull(rel.getField("requiredStr", false, false));
+  }
+
+  @Test
+  public void convertsNullTypeField() {
+    // Schema with a field whose type is just "null"
+    String schemaString =
+        "{\"type\":\"record\",\"name\":\"R\",\"namespace\":\"ns\",\"fields\":["
+            + "{\"name\":\"nullField\",\"type\":\"null\"},"
+            + "{\"name\":\"strField\",\"type\":\"string\"}]}";
+    Schema avroSchema = (new Schema.Parser()).parse(schemaString);
+    RelDataType rel = AvroConverter.rel(avroSchema);
+    assertNotNull(rel);
+    // The null-typed field should be filtered out (existing behavior) or handled gracefully
+    assertNotNull(rel.getField("strField", false, false));
+  }
+
+  @Test
+  public void convertsNullSchema() {
+    // A standalone NULL schema should not throw NPE
+    Schema nullSchema = Schema.create(Schema.Type.NULL);
+    RelDataType rel = AvroConverter.rel(nullSchema);
+    assertNotNull(rel);
+  }
+
+  @Test
+  public void handlesNullSchemaParameter() {
+    // Passing null schema should not throw NPE - should handle gracefully
+    RelDataTypeFactory typeFactory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
+    RelDataType rel = AvroConverter.rel(null, typeFactory);
+    assertNotNull(rel);
+  }
+
   @Test
   public void handlesNamespaceInNestedArrayAndMapElements() {
     RelDataTypeFactory typeFactory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sPipelineElementApi.java

@@ -69,7 +69,13 @@ List<String> getPipelineElements(V1alpha1Pipeline pipeline) {
    * Returns spec.configs of the given element, if present.
    */
   Map<String, String> getElementConfiguration(String elementYaml, String pipelineNamespace) {
-    DynamicKubernetesObject obj = Dynamics.newFromYaml(elementYaml);
+    DynamicKubernetesObject obj;
+    try {
+      obj = Dynamics.newFromYaml(elementYaml);
+    } catch (Exception e) {
+      log.error("Failed to parse element YAML for configuration extraction: {}", e.getMessage(), e);
+      return new HashMap<>();
+    }
     String name = obj.getMetadata().getName();
     String namespace = obj.getMetadata().getNamespace() == null ? pipelineNamespace : obj.getMetadata().getNamespace();
     String kind = obj.getKind();

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/status/K8sPipelineElementStatusEstimator.java

@@ -48,7 +48,19 @@ public List<K8sPipelineElementStatus> estimateStatuses(V1alpha1Pipeline pipeline
    * Estimates status of an element. If we can not retrieve it from K8s, we assume that it's not ready and not failed yet.
    */
   public K8sPipelineElementStatus estimateElementStatus(String elementYaml, String pipelineNamespace) {
-    DynamicKubernetesObject obj = Dynamics.newFromYaml(elementYaml);
+    DynamicKubernetesObject obj;
+    try {
+      obj = Dynamics.newFromYaml(elementYaml);
+    } catch (Exception e) {
+      String message = String.format("Failed to parse element YAML: %s", e.getMessage());
+      log.warn(message, e);
+      return defaultUnreadyStatusOnK8sObjectRetrievalFailure("unknown", message);
+    }
+    if (obj.getMetadata() == null) {
+      String message = "Failed to parse element YAML: null metadata";
+      log.warn(message);
+      return defaultUnreadyStatusOnK8sObjectRetrievalFailure("unknown", message);
+    }
     String name = obj.getMetadata().getName();
     String namespace = obj.getMetadata().getNamespace() == null ? pipelineNamespace : obj.getMetadata().getNamespace();
     String kind = obj.getKind();

hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sPipelineElementApiTest.java

@@ -250,6 +250,15 @@ void testGetElementConfigurationNullElement() {
         assertTrue(result.isEmpty());
     }
 
+    @Test
+    void testGetElementConfigurationMalformedYaml() {
+      // snakeyaml throws IndexOutOfBoundsException for malformed YAML.
+      // getElementConfiguration() must catch this and return an empty map rather than propagating.
+      String malformedYaml = "malformed: yaml: [unclosed";
+      Map<String, String> result = api.getElementConfiguration(malformedYaml, "default");
+      assertTrue(result.isEmpty());
+    }
+
     @Test
     void testGetElementConfigurationExceptionThrown() {
         // Given
@@ -300,4 +309,4 @@ private void setupMockForElementConfiguration(JsonObject rootJson, boolean succe
         when(mockDynamicApi.get(anyString(), anyString())).thenReturn(mockApiResponse);
         when(mockContext.dynamic(anyString(), anyString())).thenReturn(mockDynamicApi);
     }
-}
+}

hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/status/K8sPipelineElementStatusEstimatorTest.java

@@ -81,6 +81,9 @@ public class K8sPipelineElementStatusEstimatorTest {
   @BeforeEach
   void setUp() {
     estimator = new K8sPipelineElementStatusEstimator(context);
+  }
+
+  private void setUpPipelineMocks() {
     when(pipelineMetadata.getNamespace()).thenReturn("fake-namespace");
     when(pipeline.getMetadata()).thenReturn(pipelineMetadata);
     when(pipeline.getSpec()).thenReturn(pipelineSpec);
@@ -92,6 +95,7 @@ void setUp() {
 
   @Test
   void testEstimateWhenPipelineHasMultipleElements() {
+    setUpPipelineMocks();
     when(pipelineSpec.getYaml()).thenReturn(FAKE_MULTIPLE_SPECS);
     // Set up: failed to get kafka object from K8s.
     when(kafkaDynamicKubernetesApiResponse.isSuccess()).thenReturn(false);
@@ -121,6 +125,7 @@ void testEstimateWhenPipelineHasMultipleElements() {
 
   @Test
   void testEstimateWhenPipelineHasSingleElementWithK8sObjectHavingStatusFieldWithReadyInfo() {
+    setUpPipelineMocks();
     mockJobDynamicObjectWithStatusField();
     JsonElement readyElement = mock(JsonElement.class);
     when(readyElement.getAsBoolean()).thenReturn(true);
@@ -152,6 +157,7 @@ private void mockJobDynamicObjectWithStatusField() {
   @FieldSource("READY_STRINGS")
   @FieldSource("FAILED_STRINGS")
   void testEstimateWhenPipelineHasSingleElementWithK8sObjectHavingStatusFieldWithStateInfo(String state) {
+    setUpPipelineMocks();
     mockJobDynamicObjectWithStatusField();
     JsonElement stateElement = mock(JsonElement.class);
     when(stateElement.getAsString()).thenReturn(state);
@@ -169,6 +175,7 @@ void testEstimateWhenPipelineHasSingleElementWithK8sObjectHavingStatusFieldWithS
   @FieldSource("READY_STRINGS")
   @FieldSource("FAILED_STRINGS")
   void testEstimateWhenPipelineHasSingleElementWithK8sObjectHavingStatusFieldWithJobStatusStateInfo(String state) {
+    setUpPipelineMocks();
     mockJobDynamicObjectWithStatusField();
     JsonElement stateElement = mock(JsonElement.class);
     when(stateElement.getAsString()).thenReturn(state);
@@ -189,6 +196,7 @@ void testEstimateWhenPipelineHasSingleElementWithK8sObjectHavingStatusFieldWithJ
 
   @Test
   void testEstimateWhenPipelineHasSingleElementWithK8sObjectHavingNoStatusField() {
+    setUpPipelineMocks();
     when(jobDynamicObjectRawJsonObject.has("status")).thenReturn(false);
     mockDynamicObjectWithMetadata();
 
@@ -211,6 +219,7 @@ private void mockDynamicObjectWithMetadata() {
 
   @Test
   void testEstimateWhenCallToK8sFails() {
+    setUpPipelineMocks();
     when(jobDynamicKubernetesApiResponse.isSuccess()).thenReturn(false);
     List<K8sPipelineElementStatus> statuses = estimator.estimateStatuses(pipeline);
     K8sPipelineElementStatus status = Iterables.getOnlyElement(statuses);
@@ -221,6 +230,7 @@ void testEstimateWhenCallToK8sFails() {
 
   @Test
   void testEstimateWhenK8sReturnsNullObject() {
+    setUpPipelineMocks();
     when(jobDynamicKubernetesApiResponse.getObject()).thenReturn(null);
     List<K8sPipelineElementStatus> statuses = estimator.estimateStatuses(pipeline);
     K8sPipelineElementStatus status = Iterables.getOnlyElement(statuses);
@@ -229,8 +239,63 @@ void testEstimateWhenK8sReturnsNullObject() {
     assertEquals(status.getMessage(), "Returned K8s object is null or has no json");
   }
 
+  @Test
+  void testEstimateWhenElementYamlIsMalformed() {
+    // estimateElementStatus() must catch snakeyaml exceptions and return an unready status rather
+    // than propagating. Call estimateElementStatus() directly to avoid triggering unused @BeforeEach stubs.
+    K8sPipelineElementStatus status = estimator.estimateElementStatus("malformed: yaml: [unclosed", "test-namespace");
+    assertFalse(status.isReady());
+    assertFalse(status.isFailed());
+    assertTrue(status.getMessage().contains("Failed to parse element YAML"));
+  }
+
+  @Test
+  void testEstimateWhenElementYamlTriggersScannerException() {
+    // snakeyaml throws ScannerException for YAML with invalid characters or mapping structure
+    // (e.g. a colon in an unquoted value). estimateElementStatus() must catch it and return an unready status.
+    K8sPipelineElementStatus status =
+        estimator.estimateElementStatus("key: value: with: colons: everywhere:", "test-namespace");
+    assertFalse(status.isReady());
+    assertFalse(status.isFailed());
+    assertTrue(status.getMessage().contains("Failed to parse element YAML"));
+  }
+
+  @Test
+  void testEstimateWhenElementYamlTriggersParserException() {
+    // snakeyaml throws ParserException for YAML with invalid document structure
+    // (e.g. duplicate document markers). estimateElementStatus() must catch it and return an unready status.
+    K8sPipelineElementStatus status =
+        estimator.estimateElementStatus("--- invalid\n--- also invalid\n---", "test-namespace");
+    assertFalse(status.isReady());
+    assertFalse(status.isFailed());
+    assertTrue(status.getMessage().contains("Failed to parse element YAML"));
+  }
+
+  @Test
+  void testEstimateWhenElementYamlTriggersConstructorException() {
+    // snakeyaml throws ConstructorException for YAML with type tags that cannot be constructed.
+    // estimateElementStatus() must catch it and return an unready status.
+    K8sPipelineElementStatus status =
+        estimator.estimateElementStatus("!!java.util.Date 2021-01-01", "test-namespace");
+    assertFalse(status.isReady());
+    assertFalse(status.isFailed());
+    assertTrue(status.getMessage().contains("Failed to parse element YAML"));
+  }
+
+  @Test
+  void testEstimateWhenElementYamlHasNullMetadata() {
+    // A YAML with no 'metadata' field produces a DynamicKubernetesObject with null metadata.
+    // estimateElementStatus() must guard against this and return an unready status.
+    K8sPipelineElementStatus status =
+        estimator.estimateElementStatus("apiVersion: v1\nkind: ConfigMap", "test-namespace");
+    assertFalse(status.isReady());
+    assertFalse(status.isFailed());
+    assertTrue(status.getMessage().contains("null metadata"));
+  }
+
   @Test
   void testEstimateWhenPipelineHasSingleElementWithK8sObjectHavingNoRawJson() {
+    setUpPipelineMocks();
     when(jobDynamicKubernetesApiResponse.getObject()).thenReturn(jobDynamicObject);
     when(jobDynamicObject.getRaw()).thenReturn(null);
     List<K8sPipelineElementStatus> statuses = estimator.estimateStatuses(pipeline);

hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/Operator.java

@@ -177,10 +177,24 @@ public void apply(String yaml, KubernetesObject owner) throws ApiException {
   }
 
   public boolean isReady(String yaml) {
-    DynamicKubernetesObject obj = Dynamics.newFromYaml(yaml);
+    DynamicKubernetesObject obj;
+    try {
+      obj = Dynamics.newFromYaml(yaml);
+    } catch (Exception e) {
+      log.warn("Failed to parse YAML in isReady check: {}", e.getMessage());
+      return false;
+    }
+    if (obj.getMetadata() == null) {
+      log.warn("Failed to check isReady: parsed YAML has null metadata.");
+      return false;
+    }
     String namespace = obj.getMetadata().getNamespace();
     String name = obj.getMetadata().getName();
     String kind = obj.getKind();
+    if (namespace == null) {
+      log.warn("Failed to check isReady {}/{}: namespace is null.", kind, name);
+      return false;
+    }
     try {
       KubernetesApiResponse<DynamicKubernetesObject> existing = apiFor(obj).get(namespace, name);
       existing.onFailure((code, status) -> log.warn("Failed to fetch {}/{}: {}.", kind, name, status.getMessage()));
@@ -239,10 +253,24 @@ public static boolean isReady(DynamicKubernetesObject obj) {
   }
 
   public boolean isFailed(String yaml) {
-    DynamicKubernetesObject obj = Dynamics.newFromYaml(yaml);
+    DynamicKubernetesObject obj;
+    try {
+      obj = Dynamics.newFromYaml(yaml);
+    } catch (Exception e) {
+      log.warn("Failed to parse YAML in isFailed check: {}", e.getMessage());
+      return false;
+    }
+    if (obj.getMetadata() == null) {
+      log.warn("Failed to check isFailed: parsed YAML has null metadata.");
+      return false;
+    }
     String namespace = obj.getMetadata().getNamespace();
     String name = obj.getMetadata().getName();
     String kind = obj.getKind();
+    if (namespace == null) {
+      log.warn("Failed to check isFailed {}/{}: namespace is null.", kind, name);
+      return false;
+    }
     try {
       KubernetesApiResponse<DynamicKubernetesObject> existing = apiFor(obj).get(namespace, name);
       existing.onFailure((code, status) -> log.warn("Failed to fetch {}/{}: {}.", kind, name, status.getMessage()));

hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/subscription/SubscriptionReconciler.java

@@ -144,8 +144,15 @@ public Result reconcile(Request request) {
           status.setFailed(null);
           status.setMessage("Planned.");
         } catch (Exception e) {
-          log.error("Encountered error when planning a pipeline for {}/{} with SQL `{}`.", kind, name,
-              object.getSpec().getSql(), e);
+          // SqlValidatorException and ValidationException indicate bad user SQL (client error).
+          // Log at WARN level to avoid false server-error alerts.
+          if (isSqlValidationError(e)) {
+            log.warn("SQL validation error when planning a pipeline for {}/{} with SQL `{}`: {}", kind, name,
+                object.getSpec().getSql(), e.getMessage());
+          } else {
+            log.error("Encountered error when planning a pipeline for {}/{} with SQL `{}`.", kind, name,
+                object.getSpec().getSql(), e);
+          }
 
           // Mark the Subscription as failed.
           status.setFailed(true);
@@ -233,7 +240,17 @@ private static boolean diverged(V1alpha1SubscriptionSpec spec, V1alpha1Subscript
 
   // Fetch attributes from downstream controllers
   private Map<String, String> fetchAttributes(String yaml) {
-    DynamicKubernetesObject obj = Dynamics.newFromYaml(yaml);
+    DynamicKubernetesObject obj;
+    try {
+      obj = Dynamics.newFromYaml(yaml);
+    } catch (Exception e) {
+      log.warn("Failed to parse YAML in fetchAttributes: {}", e.getMessage());
+      return Collections.emptyMap();
+    }
+    if (obj.getMetadata() == null) {
+      log.warn("Failed to fetch attributes: parsed YAML has null metadata.");
+      return Collections.emptyMap();
+    }
     String namespace = obj.getMetadata().getNamespace();
     String name = obj.getMetadata().getName();
     String kind = obj.getKind();
@@ -313,5 +330,19 @@ private static Map<String, String> map(Map<String, String> m) {
       return m;
     }
   }
+
+  // Checks whether an exception represents a SQL validation error (i.e., bad user SQL),
+  // as opposed to a server-side infrastructure failure.
+  static boolean isSqlValidationError(Throwable e) {
+    Throwable x = e;
+    while (x != null) {
+      String className = x.getClass().getName();
+      if (className.contains("SqlValidatorException") || className.contains("ValidationException")) {
+        return true;
+      }
+      x = x.getCause();
+    }
+    return false;
+  }
 }