Clean up several subscription artifacts (#150)

* Fix AvroConverter to match flink key name as now supplied by ConnectionService

* Remove duplicate Subscription generated files - normalize to k8s models

* Remove HoptimatorOperatorApp and normalize under PipelineOperatorApp w/ new ControllerProvider using K8sContext

Author: jogrogan

deploy/subscriptions.crd.yaml

@@ -22,8 +22,8 @@
 /** Converts between Avro and Calcite's RelDataType */
 public final class AvroConverter {
 
-  private static final String KEY_OPTION = "keys";
-  private static final String KEY_PREFIX_OPTION = "keyPrefix";
+  private static final String KEY_OPTION = "key.fields";
+  private static final String KEY_PREFIX_OPTION = "key.fields-prefix";
   private static final String PRIMITIVE_KEY = "KEY";
 
   private AvroConverter() {

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

@@ -131,8 +131,8 @@ public void testAvroKeyPayloadSchemaValidKeyOptions() {
         List.of("KEY_field1", "field2"));
 
     Map<String, String> keyOptions = Map.of(
-        "keys", "KEY_field1",
-        "keyPrefix", "KEY_"
+        "key.fields", "KEY_field1",
+        "key.fields-prefix", "KEY_"
     );
     Pair<Schema, Schema> result = AvroConverter.avroKeyPayloadSchema("namespace", "keySchema", "payloadSchema", dataType, keyOptions);
 
@@ -160,7 +160,7 @@ public void testAvroKeyPayloadSchemaPrimitiveKey() {
         List.of("field1", "KEY"));
 
     Map<String, String> keyOptions = Map.of(
-        "keys", "KEY"
+        "key.fields", "KEY"
     );
     Pair<Schema, Schema> result = AvroConverter.avroKeyPayloadSchema("namespace", "keySchema", "payloadSchema", dataType, keyOptions);
 

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

@@ -33,6 +33,7 @@
 public final class K8sContext {
   public static final String DEFAULT_NAMESPACE = "default";
   public static final String NAMESPACE_KEY = "k8s.namespace";
+  public static final String WATCH_NAMESPACE_KEY = "k8s.watch.namespace";
   public static final String SERVER_KEY = "k8s.server";
   public static final String USER_KEY = "k8s.user";
   public static final String KUBECONFIG_KEY = "k8s.kubeconfig";
@@ -44,17 +45,19 @@ public final class K8sContext {
   public static final String SSL_TRUSTSTORE_LOCATION_KEY = "k8s.ssl.truststore.location";
 
   private final String namespace;
+  private final String watchNamespace;
   private final String clientInfo;
   private final ApiClient apiClient;
   private final SharedInformerFactory informerFactory;
   private final V1OwnerReference ownerReference;
   private final Map<String, String> labels;
   private final HoptimatorConnection connection;
 
-  private K8sContext(String namespace, String clientInfo, ApiClient apiClient,
+  private K8sContext(String namespace, String watchNamespace, String clientInfo, ApiClient apiClient,
       SharedInformerFactory informerFactory, V1OwnerReference ownerReference, Map<String, String> labels,
       HoptimatorConnection connection) {
     this.namespace = namespace;
+    this.watchNamespace = watchNamespace;
     this.clientInfo = clientInfo;
     this.apiClient = apiClient;
     this.informerFactory = informerFactory;
@@ -75,6 +78,7 @@ public static K8sContext create(Connection connection) {
     } else {
       namespace = getPodNamespace();
     }
+    String watchNamespace = connectionProperties.getProperty(WATCH_NAMESPACE_KEY);
     String kubeconfig = connectionProperties.getProperty(KUBECONFIG_KEY);
     String server = connectionProperties.getProperty(SERVER_KEY);
     String user = connectionProperties.getProperty(USER_KEY);
@@ -142,19 +146,23 @@ public static K8sContext create(Connection connection) {
       }
     }
 
-    return new K8sContext(namespace, info, apiClient, new SharedInformerFactory(apiClient),
+    if (watchNamespace == null) {
+      watchNamespace = "";
+    }
+
+    return new K8sContext(namespace, watchNamespace, info, apiClient, new SharedInformerFactory(apiClient),
         null, Collections.emptyMap(), hoptimatorConnection);
   }
 
   public K8sContext withOwner(V1OwnerReference owner) {
-    return new K8sContext(namespace, clientInfo + " Owner is " + owner.getName() + ".", apiClient,
+    return new K8sContext(namespace, watchNamespace, clientInfo + " Owner is " + owner.getName() + ".", apiClient,
         informerFactory, owner, labels, connection);
   }
 
   public K8sContext withLabel(String key, String value) {
     Map<String, String> newLabels = new HashMap<>(labels);
     newLabels.put(key, value);
-    return new K8sContext(namespace, clientInfo + " Label " + key + "=" + value + ".", apiClient,
+    return new K8sContext(namespace, watchNamespace, clientInfo + " Label " + key + "=" + value + ".", apiClient,
         informerFactory, ownerReference, newLabels, connection);
   }
 
@@ -166,10 +174,19 @@ public String namespace() {
     return namespace;
   }
 
+  public String watchNamespace() {
+    return watchNamespace;
+  }
+
   public SharedInformerFactory informerFactory() {
     return informerFactory;
   }
 
+  public <T extends KubernetesObject, U extends KubernetesListObject> void registerInformer(
+      K8sApiEndpoint<T, U> endpoint, Duration resyncPeriod) {
+    registerInformer(endpoint, resyncPeriod, watchNamespace);
+  }
+
   public <T extends KubernetesObject, U extends KubernetesListObject> void registerInformer(
       K8sApiEndpoint<T, U> endpoint, Duration resyncPeriod, String watchNamespace) {
     informerFactory.sharedIndexInformerFor(generic(endpoint), endpoint.elementType(), resyncPeriod.toMillis(), watchNamespace);

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

@@ -6,6 +6,7 @@ plugins {
 
 dependencies {
   implementation project(':hoptimator-k8s')
+  implementation project(':hoptimator-util')
   implementation project(':hoptimator-operator')
   implementation project(':hoptimator-models')      // <-- marked for deletion
   implementation libs.calcite.core

hoptimator-kafka-controller/build.gradle

@@ -1,5 +1,8 @@
 package com.linkedin.hoptimator.operator.kafka;
 
+import com.linkedin.hoptimator.k8s.K8sApiEndpoint;
+import com.linkedin.hoptimator.k8s.K8sContext;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 
@@ -12,31 +15,34 @@
 import com.linkedin.hoptimator.models.V1alpha1KafkaTopic;
 import com.linkedin.hoptimator.models.V1alpha1KafkaTopicList;
 import com.linkedin.hoptimator.operator.ControllerProvider;
-import com.linkedin.hoptimator.operator.Operator;
 
 
 /** Provides a Controller plugin for KafkaTopics. */
 public class KafkaControllerProvider implements ControllerProvider {
 
-  @Override
-  public Collection<Controller> controllers(Operator operator) {
-    operator.registerApi("KafkaTopic", "kafkatopic", "kafkatopics", "hoptimator.linkedin.com", "v1alpha1",
-        V1alpha1KafkaTopic.class, V1alpha1KafkaTopicList.class);
+  public static final K8sApiEndpoint<V1alpha1KafkaTopic, V1alpha1KafkaTopicList> KAFKA_TOPICS =
+      new K8sApiEndpoint<>("KafkaTopic", "hoptimator.linkedin.com", "v1alpha1", "kafkatopics", false,
+          V1alpha1KafkaTopic.class, V1alpha1KafkaTopicList.class);
+  public static final K8sApiEndpoint<V1alpha1Acl, V1alpha1AclList> ACLS =
+      new K8sApiEndpoint<>("Acl", "hoptimator.linkedin.com", "v1alpha1", "acls", false,
+          V1alpha1Acl.class, V1alpha1AclList.class);
 
+  @Override
+  public Collection<Controller> controllers(K8sContext context) {
+    context.registerInformer(KAFKA_TOPICS, Duration.ofMinutes(5));
     // N.B. this shared CRD may be re-registered by other ControllerProviders
-    operator.registerApi("Acl", "acl", "acls", "hoptimator.linkedin.com", "v1alpha1", V1alpha1Acl.class,
-        V1alpha1AclList.class);
+    context.registerInformer(ACLS, Duration.ofMinutes(5));
 
-    Reconciler topicReconciler = new KafkaTopicReconciler(operator);
-    Controller topicController = ControllerBuilder.defaultBuilder(operator.informerFactory())
+    Reconciler topicReconciler = new KafkaTopicReconciler(context);
+    Controller topicController = ControllerBuilder.defaultBuilder(context.informerFactory())
         .withReconciler(topicReconciler)
         .withName("kafka-topic-controller")
         .withWorkerCount(1)
         .watch(x -> ControllerBuilder.controllerWatchBuilder(V1alpha1KafkaTopic.class, x).build())
         .build();
 
-    Reconciler topicAclReconciler = new KafkaTopicAclReconciler(operator);
-    Controller topicAclController = ControllerBuilder.defaultBuilder(operator.informerFactory())
+    Reconciler topicAclReconciler = new KafkaTopicAclReconciler(context);
+    Controller topicAclController = ControllerBuilder.defaultBuilder(context.informerFactory())
         .withReconciler(topicAclReconciler)
         .withName("kafka-topic-acl-controller")
         .withWorkerCount(1)

hoptimator-kafka-controller/src/main/java/com/linkedin/hoptimator/operator/kafka/KafkaControllerProvider.java

@@ -1,5 +1,11 @@
 package com.linkedin.hoptimator.operator.kafka;
 
+import com.linkedin.hoptimator.k8s.K8sApi;
+import com.linkedin.hoptimator.k8s.K8sContext;
+import com.linkedin.hoptimator.models.V1alpha1AclList;
+import com.linkedin.hoptimator.models.V1alpha1KafkaTopicList;
+import java.sql.SQLException;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -25,18 +31,25 @@
 import com.linkedin.hoptimator.models.V1alpha1AclSpec;
 import com.linkedin.hoptimator.models.V1alpha1KafkaTopic;
 import com.linkedin.hoptimator.operator.ConfigAssembler;
-import com.linkedin.hoptimator.operator.Operator;
 
 
 public class KafkaTopicAclReconciler implements Reconciler {
   private static final Logger log = LoggerFactory.getLogger(KafkaTopicAclReconciler.class);
-  private static final String ACL = "hoptimator.linkedin.com/v1alpha1/Acl";
-  private static final String KAFKATOPIC = "hoptimator.linkedin.com/v1alpha1/KafkaTopic";
 
-  private final Operator operator;
+  private final K8sContext context;
+  private final K8sApi<V1alpha1KafkaTopic, V1alpha1KafkaTopicList> kafkaTopicApi;
+  private final K8sApi<V1alpha1Acl, V1alpha1AclList> aclApi;
 
-  public KafkaTopicAclReconciler(Operator operator) {
-    this.operator = operator;
+  public KafkaTopicAclReconciler(K8sContext context) {
+    this(context, new K8sApi<>(context, KafkaControllerProvider.KAFKA_TOPICS),
+        new K8sApi<>(context, KafkaControllerProvider.ACLS));
+  }
+
+  KafkaTopicAclReconciler(K8sContext context, K8sApi<V1alpha1KafkaTopic, V1alpha1KafkaTopicList> kafkaTopicApi,
+      K8sApi<V1alpha1Acl, V1alpha1AclList> aclApi) {
+    this.context = context;
+    this.kafkaTopicApi = kafkaTopicApi;
+    this.aclApi = aclApi;
   }
 
   @Override
@@ -46,11 +59,15 @@ public Result reconcile(Request request) {
     String namespace = request.getNamespace();
 
     try {
-      V1alpha1Acl object = operator.fetch(ACL, namespace, name);
-
-      if (object == null) {
-        log.info("Object {}/{} deleted. Skipping.", namespace, name);
-        return new Result(false);
+      V1alpha1Acl object;
+      try {
+        object = aclApi.get(namespace, name);
+      } catch (SQLException e) {
+        if (e.getErrorCode() == 404) {
+          log.info("Object {} deleted. Skipping.", name);
+          return new Result(false);
+        }
+        throw e;
       }
 
       String targetKind = Objects.requireNonNull(object.getSpec()).getResource().getKind();
@@ -77,15 +94,15 @@ public Result reconcile(Request request) {
       String targetName = object.getSpec().getResource().getName();
       String principal = object.getSpec().getPrincipal();
 
-      V1alpha1KafkaTopic target = operator.fetch(KAFKATOPIC, namespace, targetName);
+      V1alpha1KafkaTopic target = kafkaTopicApi.get(namespace, targetName);
 
       if (target == null) {
         log.info("Target KafkaTopic {}/{} not found. Retrying.", namespace, targetName);
-        return new Result(true, operator.failureRetryDuration());
+        return new Result(true, failureRetryDuration());
       }
 
       // assemble AdminClient config
-      ConfigAssembler assembler = new ConfigAssembler(operator);
+      ConfigAssembler assembler = new ConfigAssembler(context);
       list(Objects.requireNonNull(target.getSpec()).getClientConfigs()).forEach(
           x -> assembler.addRef(namespace, Objects.requireNonNull(x.getConfigMapRef()).getName()));
       map(target.getSpec().getClientOverrides()).forEach(assembler::addOverride);
@@ -102,7 +119,7 @@ public Result reconcile(Request request) {
       }
     } catch (Exception e) {
       log.error("Encountered exception while reconciling KafkaTopic Acl {}/{}", namespace, name, e);
-      return new Result(true, operator.failureRetryDuration());
+      return new Result(true, failureRetryDuration());
     }
     log.info("Done reconciling {}/{}", namespace, name);
     return new Result(false);
@@ -115,5 +132,15 @@ private static <T> List<T> list(List<T> maybeNull) {
   private static <K, V> Map<K, V> map(Map<K, V> maybeNull) {
     return Objects.requireNonNullElse(maybeNull, Collections.emptyMap());
   }
+
+  // TODO load from configuration
+  protected Duration failureRetryDuration() {
+    return Duration.ofMinutes(5);
+  }
+
+  // TODO load from configuration
+  protected Duration pendingRetryDuration() {
+    return Duration.ofMinutes(1);
+  }
 }