[#2466] feat(catalog-kafka): Add code skeleton for Kafka catalog

api/src/main/java/com/datastrato/gravitino/Catalog.java

@@ -6,6 +6,7 @@
 
 import com.datastrato.gravitino.annotation.Evolving;
 import com.datastrato.gravitino.file.FilesetCatalog;
+import com.datastrato.gravitino.messaging.TopicCatalog;
 import com.datastrato.gravitino.rel.SupportsSchemas;
 import com.datastrato.gravitino.rel.TableCatalog;
 import java.util.Map;
@@ -90,4 +91,12 @@ default TableCatalog asTableCatalog() throws UnsupportedOperationException {
   default FilesetCatalog asFilesetCatalog() throws UnsupportedOperationException {
     throw new UnsupportedOperationException("Catalog does not support fileset operations");
   }
+
+  /**
+   * @return the {@link TopicCatalog} if the catalog supports topic operations.
+   * @throws UnsupportedOperationException if the catalog does not support topic operations.
+   */
+  default TopicCatalog asTopicCatalog() throws UnsupportedOperationException {
+    throw new UnsupportedOperationException("Catalog does not support topic operations");
+  }
 }

build.gradle.kts

@@ -673,7 +673,8 @@ tasks {
       ":catalogs:catalog-lakehouse-iceberg:copyLibAndConfig",
       ":catalogs:catalog-jdbc-mysql:copyLibAndConfig",
       ":catalogs:catalog-jdbc-postgresql:copyLibAndConfig",
-      ":catalogs:catalog-hadoop:copyLibAndConfig"
+      ":catalogs:catalog-hadoop:copyLibAndConfig",
+      "catalogs:catalog-messaging-kafka:copyLibAndConfig"
     )
   }
 

catalogs/catalog-hadoop/src/main/java/com/datastrato/gravitino/catalog/hadoop/HadoopCatalogOperations.java

@@ -494,6 +494,12 @@ public PropertiesMetadata tablePropertiesMetadata() throws UnsupportedOperationE
         "Hadoop fileset catalog doesn't support table related operations");
   }
 
+  @Override
+  public PropertiesMetadata topicPropertiesMetadata() throws UnsupportedOperationException {
+    throw new UnsupportedOperationException(
+        "Hadoop fileset catalog doesn't support topic related operations");
+  }
+
   @Override
   public PropertiesMetadata catalogPropertiesMetadata() throws UnsupportedOperationException {
     return CATALOG_PROPERTIES_METADATA;

catalogs/catalog-hive/src/main/java/com/datastrato/gravitino/catalog/hive/HiveCatalogOperations.java

@@ -1088,6 +1088,12 @@ public PropertiesMetadata filesetPropertiesMetadata() throws UnsupportedOperatio
         "Hive catalog does not support fileset properties metadata");
   }
 
+  @Override
+  public PropertiesMetadata topicPropertiesMetadata() throws UnsupportedOperationException {
+    throw new UnsupportedOperationException(
+        "Hive catalog does not support topic properties metadata");
+  }
+
   CachedClientPool getClientPool() {
     return clientPool;
   }

catalogs/catalog-jdbc-common/src/main/java/com/datastrato/gravitino/catalog/jdbc/JdbcCatalogOperations.java

@@ -511,4 +511,10 @@ public PropertiesMetadata filesetPropertiesMetadata() throws UnsupportedOperatio
     throw new UnsupportedOperationException(
         "Jdbc catalog doesn't support fileset related operations");
   }
+
+  @Override
+  public PropertiesMetadata topicPropertiesMetadata() throws UnsupportedOperationException {
+    throw new UnsupportedOperationException(
+        "Jdbc catalog doesn't support topic related operations");
+  }
 }

catalogs/catalog-lakehouse-iceberg/src/main/java/com/datastrato/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java

@@ -576,4 +576,10 @@ public PropertiesMetadata filesetPropertiesMetadata() throws UnsupportedOperatio
     throw new UnsupportedOperationException(
         "Iceberg catalog doesn't support fileset related operations");
   }
+
+  @Override
+  public PropertiesMetadata topicPropertiesMetadata() throws UnsupportedOperationException {
+    throw new UnsupportedOperationException(
+        "Iceberg catalog doesn't support topic related operations");
+  }
 }

catalogs/catalog-messaging-kafka/build.gradle.kts

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+description = "catalog-messaging-kafka"
+
+plugins {
+  `maven-publish`
+  id("java")
+  id("idea")
+}
+
+dependencies {
+  implementation(project(":api"))
+  implementation(project(":core"))
+  implementation(project(":common"))
+}
+
+tasks {
+  val runtimeJars by registering(Copy::class) {
+    from(configurations.runtimeClasspath)
+    into("build/libs")
+  }
+
+  val copyCatalogLibs by registering(Copy::class) {
+    dependsOn(jar, runtimeJars)
+    from("build/libs")
+    into("$rootDir/distribution/package/catalogs/messaging-kafka/libs")
+  }
+
+  val copyCatalogConfig by registering(Copy::class) {
+    from("src/main/resources")
+    into("$rootDir/distribution/package/catalogs/messaging-kafka/conf")
+
+    // TODO. add configuration file later on.
+
+    rename { original ->
+      if (original.endsWith(".template")) {
+        original.replace(".template", "")
+      } else {
+        original
+      }
+    }
+
+    exclude { details ->
+      details.file.isDirectory()
+    }
+  }
+
+  register("copyLibAndConfig", Copy::class) {
+    dependsOn(copyCatalogConfig, copyCatalogLibs)
+  }
+}
+
+// TODO. add test task later on.

catalogs/catalog-messaging-kafka/src/main/java/com/datastrato/gravitino/catalog/kafka/KafkaCatalog.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.catalog.kafka;
+
+import com.datastrato.gravitino.catalog.BaseCatalog;
+import com.datastrato.gravitino.catalog.CatalogOperations;
+import com.datastrato.gravitino.messaging.TopicCatalog;
+import com.datastrato.gravitino.rel.SupportsSchemas;
+import java.util.Map;
+
+/** Kafka catalog is a messaging catalog that can manage topics on the Kafka messaging system. */
+public class KafkaCatalog extends BaseCatalog<KafkaCatalog> {
+
+  @Override
+  public String shortName() {
+    return "kafka";
+  }
+
+  @Override
+  protected CatalogOperations newOps(Map<String, String> config) {
+    KafkaCatalogOperations ops = new KafkaCatalogOperations(entity());
+    ops.initialize(config);
+    return ops;
+  }
+
+  @Override
+  public SupportsSchemas asSchemas() throws UnsupportedOperationException {
+    return (KafkaCatalogOperations) ops();
+  }
+
+  @Override
+  public TopicCatalog asTopicCatalog() throws UnsupportedOperationException {
+    return (KafkaCatalogOperations) ops();
+  }
+}

catalogs/catalog-messaging-kafka/src/main/java/com/datastrato/gravitino/catalog/kafka/KafkaCatalogOperations.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.catalog.kafka;
+
+import com.datastrato.gravitino.Entity;
+import com.datastrato.gravitino.EntityStore;
+import com.datastrato.gravitino.GravitinoEnv;
+import com.datastrato.gravitino.NameIdentifier;
+import com.datastrato.gravitino.Namespace;
+import com.datastrato.gravitino.catalog.CatalogOperations;
+import com.datastrato.gravitino.catalog.PropertiesMetadata;
+import com.datastrato.gravitino.exceptions.NoSuchCatalogException;
+import com.datastrato.gravitino.exceptions.NoSuchSchemaException;
+import com.datastrato.gravitino.exceptions.NoSuchTopicException;
+import com.datastrato.gravitino.exceptions.NonEmptySchemaException;
+import com.datastrato.gravitino.exceptions.SchemaAlreadyExistsException;
+import com.datastrato.gravitino.exceptions.TopicAlreadyExistsException;
+import com.datastrato.gravitino.messaging.DataLayout;
+import com.datastrato.gravitino.messaging.Topic;
+import com.datastrato.gravitino.messaging.TopicCatalog;
+import com.datastrato.gravitino.messaging.TopicChange;
+import com.datastrato.gravitino.meta.CatalogEntity;
+import com.datastrato.gravitino.meta.SchemaEntity;
+import com.datastrato.gravitino.rel.Schema;
+import com.datastrato.gravitino.rel.SchemaChange;
+import com.datastrato.gravitino.rel.SupportsSchemas;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+public class KafkaCatalogOperations implements CatalogOperations, SupportsSchemas, TopicCatalog {
+
+  private static final KafkaCatalogPropertiesMetadata CATALOG_PROPERTIES_METADATA =
+      new KafkaCatalogPropertiesMetadata();
+  private static final KafkaSchemaPropertiesMetadata SCHEMA_PROPERTIES_METADATA =
+      new KafkaSchemaPropertiesMetadata();
+  private static final KafkaTopicPropertiesMetadata TOPIC_PROPERTIES_METADATA =
+      new KafkaTopicPropertiesMetadata();
+
+  private final CatalogEntity entity;
+  private final EntityStore store;
+
+  public KafkaCatalogOperations(CatalogEntity entity) {
+    this.entity = entity;
+    this.store = GravitinoEnv.getInstance().entityStore();
+  }
+
+  @Override
+  public void initialize(Map<String, String> config) throws RuntimeException {
+    // TODO: Implement Kafka catalog initialization, such as creating a default schema.
+  }
+
+  @Override
+  public NameIdentifier[] listTopics(Namespace namespace) throws NoSuchSchemaException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public Topic loadTopic(NameIdentifier ident) throws NoSuchTopicException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public Topic createTopic(
+      NameIdentifier ident, String comment, DataLayout dataLayout, Map<String, String> properties)
+      throws NoSuchSchemaException, TopicAlreadyExistsException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public Topic alterTopic(NameIdentifier ident, TopicChange... changes)
+      throws NoSuchTopicException, IllegalArgumentException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public boolean dropTopic(NameIdentifier ident) throws NoSuchTopicException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public NameIdentifier[] listSchemas(Namespace namespace) throws NoSuchCatalogException {
+    try {
+      List<SchemaEntity> schemas =
+          store.list(namespace, SchemaEntity.class, Entity.EntityType.SCHEMA);
+      return schemas.stream()
+          .map(s -> NameIdentifier.of(namespace, s.name()))
+          .toArray(NameIdentifier[]::new);
+    } catch (IOException e) {
+      throw new RuntimeException("Failed to list schemas under namespace " + namespace, e);
+    }
+  }
+
+  @Override
+  public Schema createSchema(NameIdentifier ident, String comment, Map<String, String> properties)
+      throws NoSuchCatalogException, SchemaAlreadyExistsException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public Schema loadSchema(NameIdentifier ident) throws NoSuchSchemaException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public Schema alterSchema(NameIdentifier ident, SchemaChange... changes)
+      throws NoSuchSchemaException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmptySchemaException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public PropertiesMetadata catalogPropertiesMetadata() throws UnsupportedOperationException {
+    return CATALOG_PROPERTIES_METADATA;
+  }
+
+  @Override
+  public PropertiesMetadata schemaPropertiesMetadata() throws UnsupportedOperationException {
+    return SCHEMA_PROPERTIES_METADATA;
+  }
+
+  @Override
+  public PropertiesMetadata topicPropertiesMetadata() throws UnsupportedOperationException {
+    return TOPIC_PROPERTIES_METADATA;
+  }
+
+  @Override
+  public void close() throws IOException {}
+
+  @Override
+  public PropertiesMetadata filesetPropertiesMetadata() throws UnsupportedOperationException {
+    throw new UnsupportedOperationException("Kafka catalog does not support fileset operations");
+  }
+
+  @Override
+  public PropertiesMetadata tablePropertiesMetadata() throws UnsupportedOperationException {
+    throw new UnsupportedOperationException("Kafka catalog does not support table operations");
+  }
+}

catalogs/catalog-messaging-kafka/src/main/java/com/datastrato/gravitino/catalog/kafka/KafkaCatalogPropertiesMetadata.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.catalog.kafka;
+
+import com.datastrato.gravitino.catalog.BaseCatalogPropertiesMetadata;
+import com.datastrato.gravitino.catalog.PropertyEntry;
+import java.util.Collections;
+import java.util.Map;
+
+public class KafkaCatalogPropertiesMetadata extends BaseCatalogPropertiesMetadata {
+
+  // The "bootstrap.servers" property specifies the Kafka broker(s) to connect to, allowing for
+  // multiple brokers by comma-separating them.
+  public static final String BOOTSTRAP_SERVERS = "bootstrap.servers";
+
+  private static final Map<String, PropertyEntry<?>> KAFKA_CATALOG_PROPERTY_ENTRIES =
+      Collections.singletonMap(
+          BOOTSTRAP_SERVERS,
+          PropertyEntry.stringRequiredPropertyEntry(
+              BOOTSTRAP_SERVERS,
+              "The Kafka broker(s) to connect to, allowing for multiple brokers by comma-separating them",
+              true /* immutable */,
+              false /* hidden */));
+
+  @Override
+  protected Map<String, PropertyEntry<?>> specificPropertyEntries() {
+    return KAFKA_CATALOG_PROPERTY_ENTRIES;
+  }
+}

catalogs/catalog-messaging-kafka/src/main/java/com/datastrato/gravitino/catalog/kafka/KafkaSchemaPropertiesMetadata.java

@@ -0,0 +1,18 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.catalog.kafka;
+
+import com.datastrato.gravitino.catalog.BasePropertiesMetadata;
+import com.datastrato.gravitino.catalog.PropertyEntry;
+import java.util.Collections;
+import java.util.Map;
+
+public class KafkaSchemaPropertiesMetadata extends BasePropertiesMetadata {
+
+  @Override
+  protected Map<String, PropertyEntry<?>> specificPropertyEntries() {
+    return Collections.emptyMap();
+  }
+}

catalogs/catalog-messaging-kafka/src/main/java/com/datastrato/gravitino/catalog/kafka/KafkaTopicPropertiesMetadata.java

@@ -0,0 +1,17 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.catalog.kafka;
+
+import com.datastrato.gravitino.catalog.BasePropertiesMetadata;
+import com.datastrato.gravitino.catalog.PropertyEntry;
+import java.util.Collections;
+import java.util.Map;
+
+public class KafkaTopicPropertiesMetadata extends BasePropertiesMetadata {
+  @Override
+  protected Map<String, PropertyEntry<?>> specificPropertyEntries() {
+    return Collections.emptyMap();
+  }
+}