Fix NPE again (#44)

Author: ryannedolan

deploy/samples/subscriptions.yaml

@@ -3,8 +3,19 @@ apiVersion: hoptimator.linkedin.com/v1alpha1
 kind: Subscription
 metadata:
   name: products
+spec:
+  sql: SELECT "quantity", "product_id" AS KEY FROM INVENTORY."products_on_hand"
+  database: RAWKAFKA
+
+---
+
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: Subscription
+metadata:
+  name: products-with-hints
 spec:
   sql: SELECT "quantity", "product_id" AS KEY FROM INVENTORY."products_on_hand"
   database: RAWKAFKA
   hints:
-    numPartitions: "2"
+    kafka.numPartitions: "7"
+

etc/integration-tests.sql

@@ -16,6 +16,9 @@ SELECT * FROM INVENTORY."products_on_hand" LIMIT 1;
 -- MySQL CDC -> Kafka
 SELECT * FROM RAWKAFKA."products" LIMIT 1;
 
+-- Same, but with hints:
+SELECT * FROM RAWKAFKA."products-with-hints" LIMIT 1;
+
 -- test insert into command
 !insert into RAWKAFKA."test-sink" SELECT AGE AS PAYLOAD, NAME AS KEY FROM DATAGEN.PERSON
 SELECT * FROM RAWKAFKA."test-sink" LIMIT 5;

hoptimator-kafka-adapter/src/main/resources/KafkaTopic.yaml.template

@@ -5,6 +5,6 @@ metadata:
   namespace: {{pipeline.namespace}}
 spec:
   topicName: {{topicName}}
-  numPartitions: {{numPartitions:null}}
+  numPartitions: {{kafka.numPartitions:null}}
   clientOverrides:
     {{clientOverrides}}

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

@@ -93,7 +93,7 @@ public Result reconcile(Request request) {
 
         // For sink resources, also expose hints.
         Resource.TemplateFactory sinkTemplateFactory = new Resource.SimpleTemplateFactory(subEnv
-          .orElse(new Resource.SimpleEnvironment(object.getSpec().getHints())));
+          .orElse(new Resource.SimpleEnvironment(map(object.getSpec().getHints()))));
 
         // Render resources related to all source tables.
         List<String> upstreamResources = pipeline.upstreamResources().stream()
@@ -284,5 +284,13 @@ public static Controller controller(Operator operator, HoptimatorPlanner.Factory
       .watch(x -> ControllerBuilder.controllerWatchBuilder(V1alpha1Subscription.class, x).build())
       .build();
   }
+
+  private static Map<String, String> map(Map<String, String> m) {
+    if (m == null) {
+      return Collections.emptyMap();
+    } else {
+      return m;
+    }
+  }
 }