Remove schema registry support from core module

README.md

@@ -109,8 +109,6 @@ The following configuration options are available:
 
 - `--bootstrap-servers`, `--bootstrap-server`: List of Kafka bootstrap servers (comma-separated) (**required**)
 
-- `--schema-registry-url`: The URL of the Schema Registry
-
 - `--kafka-config`: Kafka Streams configuration (`<String=String>[,<String=String>...]`)
 
 - `--input-topics`: List of input topics (comma-separated)
@@ -195,8 +193,6 @@ The following configuration options are available:
 
 - `--bootstrap-servers`, `--bootstrap-server`: List of Kafka bootstrap servers (comma-separated) (**required**)
 
-- `--schema-registry-url`: The URL of the Schema Registry
-
 - `--kafka-config`: Kafka producer configuration (`<String=String>[,<String=String>...]`)
 
 - `--output-topic`: The output topic

charts/producer-app-cleanup-job/templates/job.yaml

@@ -60,10 +60,6 @@ spec:
             - name: "{{ .Values.configurationEnvPrefix }}_BOOTSTRAP_SERVERS"
               value: {{ .Values.kafka.bootstrapServers | quote }}
             {{- end }}
-            {{- if hasKey .Values.kafka "schemaRegistryUrl" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
-              value: {{ .Values.kafka.schemaRegistryUrl | quote }}
-            {{- end }}
             {{- if hasKey .Values.kafka "outputTopic" }}
             - name: "{{ .Values.configurationEnvPrefix }}_OUTPUT_TOPIC"
               value: {{ .Values.kafka.outputTopic | quote }}

charts/producer-app-cleanup-job/values.yaml

@@ -18,7 +18,6 @@ files: {}
 
 kafka:
 #  bootstrapServers: "test:9092"
-#  schemaRegistryUrl: "url:1234"
   config: {}
 #    max.poll.records: 500
 #    Note that YAML may convert large integers to scientific notation. Use Strings to avoid this.

charts/producer-app/README.md

@@ -50,7 +50,6 @@ Alternatively, a YAML file that specifies the values for the parameters can be p
 | Parameter                   | Description                                                                                                  | Default |
 |-----------------------------|--------------------------------------------------------------------------------------------------------------|---------|
 | `kafka.bootstrapServers`    | Comma separated list of Kafka bootstrap servers to connect to.                                               |         |
-| `kafka.schemaRegistryUrl`   | URL of Schema Registry to connect to.                                                                        | `null`  |
 | `kafka.config`              | Configurations for your [Kafka producer app](https://kafka.apache.org/documentation/#producerconfigs).       | `{}`    |
 | `kafka.outputTopic`         | Output topic for your producer application.                                                                  |         |
 | `kafka.labeledOutputTopics` | Map of additional labeled output topics if you need to specify multiple topics with different message types. | `{}`    |

charts/producer-app/templates/_pod.yaml

@@ -52,10 +52,6 @@ spec:
       - name: "{{ .Values.configurationEnvPrefix }}_BOOTSTRAP_SERVERS"
         value: {{ .Values.kafka.bootstrapServers | quote }}
         {{- end }}
-        {{- if hasKey .Values.kafka "schemaRegistryUrl" }}
-      - name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
-        value: {{ .Values.kafka.schemaRegistryUrl | quote }}
-        {{- end }}
         {{- if hasKey .Values.kafka "outputTopic" }}
       - name: "{{ .Values.configurationEnvPrefix }}_OUTPUT_TOPIC"
         value: {{ .Values.kafka.outputTopic | quote }}

charts/producer-app/values.yaml

@@ -51,7 +51,6 @@ resources:
 
 kafka:
 #  bootstrapServers: "test:9092"
-#  schemaRegistryUrl: "url:1234"
   config: {}
 #    max.poll.records: 500
 #    Note that YAML may convert large integers to scientific notation. Use Strings to avoid this.

charts/streams-app-cleanup-job/templates/job.yaml

@@ -64,10 +64,6 @@ spec:
             - name: "{{ .Values.configurationEnvPrefix }}_BOOTSTRAP_SERVERS"
               value: {{ .Values.kafka.bootstrapServers | quote }}
             {{- end }}
-            {{- if hasKey .Values.kafka "schemaRegistryUrl" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
-              value: {{ .Values.kafka.schemaRegistryUrl | quote }}
-            {{- end }}
             {{- if and (hasKey .Values.kafka "inputTopics") (.Values.kafka.inputTopics) }}
             - name: "{{ .Values.configurationEnvPrefix }}_INPUT_TOPICS"
               value: {{ .Values.kafka.inputTopics | join "," | quote }}

charts/streams-app-cleanup-job/values.yaml

@@ -18,7 +18,6 @@ files: {}
 
 kafka:
   # bootstrapServers: "test:9092"
-  # schemaRegistryUrl: "url:1234"
   config: {}
   #   max.poll.records: 500
   #   Note that YAML may convert large integers to scientific notation. Use Strings to avoid this.

charts/streams-app/README.md

@@ -54,7 +54,6 @@ Alternatively, a YAML file that specifies the values for the parameters can be p
 | Parameter                    | Description                                                                                                                                                                      | Default |
 |------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------|
 | `kafka.bootstrapServers`     | Comma separated list of Kafka bootstrap servers to connect to.                                                                                                                   |         |
-| `kafka.schemaRegistryUrl`    | URL of Schema Registry to connect to.                                                                                                                                            | `null`  |
 | `kafka.staticMembership`     | Whether to use [Kafka Static Group Membership](https://cwiki.apache.org/confluence/display/KAFKA/KIP-345%3A+Introduce+static+membership+protocol+to+reduce+consumer+rebalances). | `false` |
 | `kafka.config`               | Configurations for your [Kafka Streams app](https://kafka.apache.org/documentation/#streamsconfigs).                                                                             | `{}`    |
 | `kafka.inputTopics`          | List of input topics for your streams application.                                                                                                                               | `[]`    |

charts/streams-app/templates/deployment.yaml

@@ -107,10 +107,6 @@ spec:
             - name: "{{ .Values.configurationEnvPrefix }}_BOOTSTRAP_SERVERS"
               value: {{ .Values.kafka.bootstrapServers | quote }}
           {{- end }}
-          {{- if hasKey .Values.kafka "schemaRegistryUrl" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
-              value: {{ .Values.kafka.schemaRegistryUrl | quote }}
-          {{- end }}
           {{- if and (hasKey .Values.kafka "inputTopics") (.Values.kafka.inputTopics) }}
             - name: "{{ .Values.configurationEnvPrefix }}_INPUT_TOPICS"
               value: {{ .Values.kafka.inputTopics | join "," | quote }}

charts/streams-app/values.yaml

@@ -26,7 +26,6 @@ resources:
 
 kafka:
   # bootstrapServers: "test:9092"
-  # schemaRegistryUrl: "url:1234"
   staticMembership: false
   config: {}
   #   max.poll.records: 500

settings.gradle.kts

@@ -10,6 +10,7 @@ include(
         ":streams-bootstrap-core",
         ":streams-bootstrap-test",
         ":streams-bootstrap-large-messages",
+        ":streams-bootstrap-schema-registry",
         ":streams-bootstrap-cli",
         ":streams-bootstrap-cli-test",
         ":streams-bootstrap-bom",

streams-bootstrap-bom/build.gradle.kts

@@ -9,6 +9,7 @@ dependencies {
         api(project(":streams-bootstrap-core"))
         api(project(":streams-bootstrap-cli"))
         api(project(":streams-bootstrap-large-messages"))
+        api(project(":streams-bootstrap-schema-registry"))
         api(project(":streams-bootstrap-test"))
         api(project(":streams-bootstrap-cli-test"))
     }

streams-bootstrap-cli-test/src/main/java/com/bakdata/kafka/TestApplicationRunner.java

@@ -235,11 +235,16 @@ public KafkaTestClient newTestClient() {
      */
     public void configure(final KafkaApplication<?, ?, ?, ?, ?, ?, ?, ?> app) {
         app.setBootstrapServers(this.bootstrapServers);
-        final Map<String, String> mergedConfig = merge(app.getKafkaConfig(), this.kafkaConfig);
+        final Map<String, String> localConfig = this.getLocalConfig();
+        final Map<String, String> mergedConfig = merge(app.getKafkaConfig(), localConfig);
         app.setKafkaConfig(mergedConfig);
-        if (this.schemaRegistry != null) {
-            app.setSchemaRegistryUrl(this.schemaRegistry.getSchemaRegistryUrl());
+    }
+
+    private Map<String, String> getLocalConfig() {
+        if (this.schemaRegistry == null) {
+            return this.kafkaConfig;
         }
+        return merge(this.kafkaConfig, this.schemaRegistry.getKafkaProperties());
     }
 
     private void prepareExecution(final KafkaApplication<?, ?, ?, ?, ?, ?, ?, ?> app) {
@@ -248,10 +253,8 @@ private void prepareExecution(final KafkaApplication<?, ?, ?, ?, ?, ?, ?, ?> app
     }
 
     private RuntimeConfiguration createRuntimeConfiguration() {
-        final RuntimeConfiguration configuration = RuntimeConfiguration.create(this.bootstrapServers)
-                .with(this.kafkaConfig);
-        return this.schemaRegistry == null ? configuration
-                : configuration.withSchemaRegistryUrl(this.schemaRegistry.getSchemaRegistryUrl());
+        return RuntimeConfiguration.create(this.bootstrapServers)
+                .with(this.getLocalConfig());
     }
 
     private String[] setupArgs(final String[] args, final Iterable<String> command) {

streams-bootstrap-cli/src/main/java/com/bakdata/kafka/KafkaApplication.java

@@ -52,7 +52,6 @@
  *     <li>{@link #bootstrapServers}</li>
  *     <li>{@link #outputTopic}</li>
  *     <li>{@link #labeledOutputTopics}</li>
- *     <li>{@link #schemaRegistryUrl}</li>
  *     <li>{@link #kafkaConfig}</li>
  * </ul>
  * To implement your Kafka application inherit from this class and add your custom options. Run it by creating an
@@ -90,8 +89,6 @@ public abstract class KafkaApplication<R extends Runner, CR extends CleanUpRunne
     @CommandLine.Option(names = {"--bootstrap-servers", "--bootstrap-server"}, required = true,
             description = "Kafka bootstrap servers to connect to")
     private String bootstrapServers;
-    @CommandLine.Option(names = "--schema-registry-url", description = "URL of Schema Registry")
-    private String schemaRegistryUrl;
     @CommandLine.Option(names = "--kafka-config", split = ",", description = "Additional Kafka properties")
     private Map<String, String> kafkaConfig = emptyMap();
 
@@ -191,10 +188,8 @@ public void run() {
     }
 
     public RuntimeConfiguration getRuntimeConfiguration() {
-        final RuntimeConfiguration configuration = RuntimeConfiguration.create(this.bootstrapServers)
+        return RuntimeConfiguration.create(this.bootstrapServers)
                 .with(this.kafkaConfig);
-        return this.schemaRegistryUrl == null ? configuration
-                : configuration.withSchemaRegistryUrl(this.schemaRegistryUrl);
     }
 
     /**

streams-bootstrap-cli/src/test/java/com/bakdata/kafka/streams/KafkaStreamsApplicationCliTest.java

@@ -365,7 +365,6 @@ public void run() {
         }) {
             app.startApplicationWithoutExit(new String[]{
                     "--bootstrap-server", "bootstrap-servers",
-                    "--schema-registry-url", "schema-registry",
                     "--input-topics", "input1,input2",
                     "--labeled-input-topics", "label1=input3,label2=input4;input5",
                     "--input-pattern", ".*",
@@ -375,7 +374,6 @@ public void run() {
                     "--kafka-config", "foo=1,bar=2",
             });
             assertThat(app.getBootstrapServers()).isEqualTo("bootstrap-servers");
-            assertThat(app.getSchemaRegistryUrl()).isEqualTo("schema-registry");
             assertThat(app.getInputTopics()).containsExactly("input1", "input2");
             assertThat(app.getLabeledInputTopics())
                     .hasSize(2)

streams-bootstrap-core/build.gradle.kts

@@ -12,14 +12,6 @@ dependencies {
 
     api(libs.kafka.streams)
     api(libs.kafka.clients)
-    implementation(libs.kafka.schema.serializer) {
-        exclude(group = "org.apache.kafka", module = "kafka-clients") // force usage of OSS kafka-clients
-        exclude(group = "org.slf4j", module = "slf4j-api") // Conflict with 2.x when used as dependency
-    }
-    api(libs.kafka.schema.registry.client) {
-        exclude(group = "org.apache.kafka", module = "kafka-clients") // force usage of OSS kafka-clients
-        exclude(group = "org.slf4j", module = "slf4j-api") // Conflict with 2.x when used as dependency
-    }
     implementation(libs.slf4j)
     implementation(libs.jool)
     implementation(libs.resilience4j.retry)
@@ -34,6 +26,7 @@ dependencies {
     testImplementation(libs.mockito.junit)
 
     testImplementation(testFixtures(project(":streams-bootstrap-test")))
+    testImplementation(project(":streams-bootstrap-schema-registry"))
     testImplementation(libs.kafka.streams.avro.serde) {
         exclude(group = "org.apache.kafka", module = "kafka-clients") // force usage of OSS kafka-clients
     }

streams-bootstrap-core/src/main/java/com/bakdata/kafka/RuntimeConfiguration.java

@@ -24,7 +24,6 @@
 
 package com.bakdata.kafka;
 
-import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.Collection;
@@ -40,13 +39,12 @@
 import org.apache.kafka.streams.StreamsConfig;
 
 /**
- * Runtime configuration to connect to Kafka infrastructure, e.g., bootstrap servers and schema registry.
+ * Runtime configuration to connect to Kafka infrastructure, e.g., bootstrap servers.
  */
 @RequiredArgsConstructor(access = AccessLevel.PRIVATE)
 public final class RuntimeConfiguration {
     private static final Set<String> PROVIDED_PROPERTIES = Set.of(
-            CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
-            AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG
+            CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG
     );
     private final @NonNull Map<String, Object> properties;
 
@@ -71,16 +69,6 @@ public RuntimeConfiguration with(final Map<String, ?> newProperties) {
         return this.withInternal(newProperties);
     }
 
-    /**
-     * Configure a schema registry for (de-)serialization.
-     *
-     * @param schemaRegistryUrl schema registry url
-     * @return a copy of this runtime configuration with configured schema registry
-     */
-    public RuntimeConfiguration withSchemaRegistryUrl(final String schemaRegistryUrl) {
-        return this.withInternal(Map.of(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl));
-    }
-
     /**
      * Configure {@link StreamsConfig#STATE_DIR_CONFIG} for Kafka Streams. Useful for testing
      *

streams-bootstrap-core/src/main/java/com/bakdata/kafka/admin/AdminClientX.java

@@ -24,14 +24,8 @@
 
 package com.bakdata.kafka.admin;
 
-import static com.bakdata.kafka.SchemaRegistryAppUtils.createSchemaRegistryClient;
-
-import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
-import java.io.IOException;
-import java.io.UncheckedIOException;
 import java.time.Duration;
 import java.util.Map;
-import java.util.Optional;
 import lombok.AccessLevel;
 import lombok.Builder;
 import lombok.NonNull;
@@ -49,7 +43,6 @@ public final class AdminClientX implements AutoCloseable {
 
     private static final Duration ADMIN_TIMEOUT = Duration.ofSeconds(10L);
     private final @NonNull Admin adminClient;
-    private final SchemaRegistryClient schemaRegistryClient;
     private final @NonNull Timeout timeout;
 
     /**
@@ -74,10 +67,8 @@ public static AdminClientX create(@NonNull final Map<String, Object> properties,
                     String.format("%s must be specified in properties", AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG));
         }
         final Admin adminClient = AdminClient.create(properties);
-        final SchemaRegistryClient schemaRegistryClient = createSchemaRegistryClient(properties).orElse(null);
         return builder()
                 .adminClient(adminClient)
-                .schemaRegistryClient(schemaRegistryClient)
                 .timeout(new Timeout(timeout))
                 .build();
     }
@@ -91,15 +82,6 @@ public Admin admin() {
         return new PooledAdmin(this.adminClient);
     }
 
-    /**
-     * Create a {@link SchemaRegistryClient} if schema registry is configured.
-     * @return schema registry client
-     */
-    public Optional<SchemaRegistryClient> schemaRegistry() {
-        return Optional.ofNullable(this.schemaRegistryClient)
-                .map(PooledSchemaRegistryClient::new);
-    }
-
     /**
      * Create a {@link TopicsClient} to perform topic-related administrative actions.
      * @return topic client
@@ -119,13 +101,6 @@ public ConsumerGroupsClient consumerGroups() {
     @Override
     public void close() {
         this.adminClient.close();
-        if (this.schemaRegistryClient != null) {
-            try {
-                this.schemaRegistryClient.close();
-            } catch (final IOException e) {
-                throw new UncheckedIOException("Error closing schema registry client", e);
-            }
-        }
     }
 
     @RequiredArgsConstructor
@@ -143,15 +118,4 @@ public void close() {
             // do nothing
         }
     }
-
-    @RequiredArgsConstructor
-    private static class PooledSchemaRegistryClient implements SchemaRegistryClient {
-        @Delegate(excludes = AutoCloseable.class)
-        private final @NonNull SchemaRegistryClient schemaRegistryClient;
-
-        @Override
-        public void close() {
-            // do nothing
-        }
-    }
 }

streams-bootstrap-core/src/main/java/com/bakdata/kafka/producer/ProducerCleanUpRunner.java

@@ -25,7 +25,6 @@
 package com.bakdata.kafka.producer;
 
 import com.bakdata.kafka.CleanUpRunner;
-import com.bakdata.kafka.SchemaRegistryAppUtils;
 import com.bakdata.kafka.admin.AdminClientX;
 import java.util.Map;
 import lombok.AccessLevel;
@@ -68,8 +67,6 @@ public static ProducerCleanUpRunner create(@NonNull final ProducerTopicConfig to
     public static ProducerCleanUpRunner create(@NonNull final ProducerTopicConfig topics,
             @NonNull final Map<String, Object> kafkaProperties,
             @NonNull final ProducerCleanUpConfiguration configuration) {
-        SchemaRegistryAppUtils.createTopicHook(kafkaProperties)
-                .ifPresent(configuration::registerTopicHook);
         return new ProducerCleanUpRunner(topics, kafkaProperties, configuration);
     }
 

streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/StreamsCleanUpRunner.java

@@ -26,7 +26,6 @@
 
 import com.bakdata.kafka.CleanUpException;
 import com.bakdata.kafka.CleanUpRunner;
-import com.bakdata.kafka.SchemaRegistryAppUtils;
 import com.bakdata.kafka.admin.AdminClientX;
 import com.bakdata.kafka.admin.ConsumerGroupsClient;
 import com.bakdata.kafka.util.TopologyInformation;
@@ -85,8 +84,6 @@ public static StreamsCleanUpRunner create(final @NonNull Topology topology,
             final @NonNull StreamsConfig streamsConfig, final @NonNull StreamsCleanUpConfiguration configuration) {
         final StreamsConfigX config = new StreamsConfigX(streamsConfig);
         final TopologyInformation topologyInformation = new TopologyInformation(topology, config.getAppId());
-        SchemaRegistryAppUtils.createTopicHook(config.getKafkaProperties())
-                .ifPresent(configuration::registerTopicHook);
         return new StreamsCleanUpRunner(topologyInformation, topology, config, configuration);
     }