Fix PipelineReconciler namespace (#130)

* Fix PipelineReconciler namespace

* Fix various checkstyle complaints

Author: jogrogan

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/CompatibilityValidatorBase.java

@@ -12,7 +12,7 @@ abstract class CompatibilityValidatorBase implements Validator {
 
   private final SchemaPlus schema;
 
-  public CompatibilityValidatorBase(SchemaPlus schema) {
+  CompatibilityValidatorBase(SchemaPlus schema) {
     this.schema = schema;
   }
 

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDdlExecutor.java

@@ -20,7 +20,6 @@
 package com.linkedin.hoptimator.jdbc;
 
 import java.io.Reader;
-import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -47,7 +46,6 @@
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
 import org.apache.calcite.sql.SqlSelect;
-import org.apache.calcite.sql.SqlUtil;
 import org.apache.calcite.sql.ddl.SqlCreateMaterializedView;
 import org.apache.calcite.sql.ddl.SqlCreateView;
 import org.apache.calcite.sql.ddl.SqlDropObject;
@@ -69,8 +67,6 @@
 import com.linkedin.hoptimator.util.DeploymentService;
 import com.linkedin.hoptimator.util.planner.PipelineRel;
 
-import static org.apache.calcite.util.Static.RESOURCE;
-
 
 public final class HoptimatorDdlExecutor extends ServerDdlExecutor {
 
@@ -135,7 +131,7 @@ public void execute(SqlCreateView create, CalcitePrepare.Context context) {
       schemaPlus.add(viewName, viewTable);
     } catch (Exception e) {
       throw new DdlException(create, e.getMessage(), e);
-    } 
+    }
   }
 
   // N.B. copy-pasted from Apache Calcite

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDriver.java

@@ -131,7 +131,7 @@ public Connection connect(String url, Properties props) throws SQLException {
         }
       }
       return hoptimatorConnection;
-    } catch (IOException|SQLTransientException e) {
+    } catch (IOException | SQLTransientException e) {
       throw new SQLTransientConnectionException("Problem loading " + url, e);
     } catch (Exception e) {
       throw new SQLNonTransientException("Problem loading " + url, e);

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

@@ -43,7 +43,7 @@ public class K8sContext {
 
   private final String namespace;
   private final String clientInfo;
-  private ApiClient apiClient;
+  private final ApiClient apiClient;
   private final SharedInformerFactory informerFactory;
   private final V1OwnerReference ownerReference;
   private final Map<String, String> labels;

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

@@ -1,16 +1,16 @@
 package com.linkedin.hoptimator.k8s;
 
 import java.sql.SQLException;
-import java.sql.SQLTransientException;
 import java.sql.SQLNonTransientException;
+import java.sql.SQLTransientException;
 import java.util.Collection;
 import java.util.Locale;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import io.kubernetes.client.common.KubernetesType;
-import io.kubernetes.client.util.generic.KubernetesApiResponse;
 import io.kubernetes.client.openapi.ApiException;
+import io.kubernetes.client.util.generic.KubernetesApiResponse;
 
 import com.linkedin.hoptimator.Sink;
 import com.linkedin.hoptimator.Source;

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

@@ -1,16 +1,13 @@
 package com.linkedin.hoptimator.k8s;
 
 import java.sql.SQLException;
-import java.sql.SQLNonTransientException;
-import java.sql.SQLTransientException;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.kubernetes.client.openapi.ApiException;
 import io.kubernetes.client.util.generic.KubernetesApiResponse;
 import io.kubernetes.client.util.generic.dynamic.DynamicKubernetesObject;
 import io.kubernetes.client.util.generic.dynamic.Dynamics;

hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/pipeline/PipelineReconciler.java

@@ -65,7 +65,7 @@ public Result reconcile(Request request) {
       boolean ready = Arrays.asList(object.getSpec().getYaml().split("\n---\n"))
           .stream()
           .filter(x -> x != null && !x.isEmpty())
-          .allMatch(x -> isReady(x));
+          .allMatch(x -> isReady(x, namespace));
 
       if (ready) {
         status.setReady(true);
@@ -115,15 +115,17 @@ public static Controller controller(K8sContext context) {
         .build();
   }
 
-  private boolean isReady(String yaml) {
+  private boolean isReady(String yaml, String pipelineNamespace) {
     DynamicKubernetesObject obj = Dynamics.newFromYaml(yaml);
-    String namespace = obj.getMetadata().getNamespace();
     String name = obj.getMetadata().getName();
+    String namespace = obj.getMetadata().getNamespace() == null ? pipelineNamespace : obj.getMetadata().getNamespace();
     String kind = obj.getKind();
     try {
       KubernetesApiResponse<DynamicKubernetesObject> existing =
           context.dynamic(obj.getApiVersion(), K8sUtils.guessPlural(obj)).get(namespace, name);
-      existing.onFailure((code, status) -> log.warn("Failed to fetch {}/{}: {}.", kind, name, status.getMessage()));
+      existing.onFailure((code, status) ->
+          log.warn("Failed to fetch {}/{} in namespace {}: {}.", kind, name, namespace, status.getMessage())
+      );
       if (!existing.isSuccess()) {
         return false;
       }
@@ -135,6 +137,7 @@ private boolean isReady(String yaml) {
         return false;
       }
     } catch (Exception e) {
+      log.error("Encountered exception while checking status of {}/{} in namespace {}", kind, name, namespace, e);
       return false;
     }
   }

hoptimator-venice/src/main/java/com/linkedin/hoptimator/venice/VeniceDriver.java

@@ -5,7 +5,6 @@
 import java.sql.SQLException;
 import java.sql.SQLNonTransientException;
 import java.sql.SQLTransientConnectionException;
-import java.sql.SQLTransientException;
 import java.util.Properties;
 import java.util.concurrent.ExecutionException;
 
@@ -57,7 +56,7 @@ public Connection connect(String url, Properties props) throws SQLException {
       schema.populate();
       rootSchema.add(CATALOG_NAME, schema);
       return connection;
-    } catch (IOException|ExecutionException|InterruptedException e) {
+    } catch (IOException | ExecutionException | InterruptedException e) {
       throw new SQLTransientConnectionException("Problem loading " + url, e);
     } catch (Exception e) {
       throw new SQLNonTransientException("Problem loading " + url, e);