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

@@ -68,10 +68,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

@@ -17,7 +17,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

@@ -49,7 +49,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

@@ -50,7 +50,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

@@ -72,10 +72,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

@@ -17,7 +17,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

@@ -53,7 +53,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

@@ -119,10 +119,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

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

settings.gradle

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

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 calling
@@ -89,8 +88,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();
 
@@ -204,9 +201,9 @@ public void run() {
     }
 
     public KafkaEndpointConfig getEndpointConfig() {
+        // We do not specify endpoint properties, those are passed via kafkaConfig
         return KafkaEndpointConfig.builder()
                 .bootstrapServers(this.bootstrapServers)
-                .schemaRegistryUrl(this.schemaRegistryUrl)
                 .build();
     }
 

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

@@ -203,7 +203,6 @@ public SerdeConfig defaultSerializationConfig() {
             }
         }, new String[]{
                 "--bootstrap-servers", "localhost:9092",
-                "--schema-registry-url", "http://localhost:8081",
                 "--input-topics", "input",
                 "--output-topic", "output",
                 "--application-id", "my-other-id"
@@ -351,7 +350,6 @@ public void run() {
         }) {
             KafkaApplication.startApplicationWithoutExit(app, 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", ".*",
@@ -361,7 +359,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-cli/src/test/java/com/bakdata/kafka/integration/RunProducerAppTest.java

@@ -91,9 +91,10 @@ public SerializerConfig defaultSerializationConfig() {
             }
         })) {
             app.setBootstrapServers(this.kafkaCluster.getBrokerList());
-            app.setSchemaRegistryUrl(this.schemaRegistryMockExtension.getUrl());
             app.setOutputTopic(output);
             app.setKafkaConfig(Map.of(
+                    AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
+                    this.schemaRegistryMockExtension.getUrl(),
                     ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000"
             ));
             app.run();

streams-bootstrap-core/build.gradle.kts

@@ -10,9 +10,6 @@ dependencies {
 
     api(group = "org.apache.kafka", name = "kafka-streams", version = kafkaVersion)
     api(group = "org.apache.kafka", name = "kafka-clients", version = kafkaVersion)
-    val confluentVersion: String by project
-    implementation(group = "io.confluent", name = "kafka-schema-serializer", version = confluentVersion)
-    api(group = "io.confluent", name = "kafka-schema-registry-client", version = confluentVersion)
     api(
         group = "org.slf4j",
         name = "slf4j-api",
@@ -33,6 +30,7 @@ dependencies {
 
     val fluentKafkaVersion: String by project
     testImplementation(project(":streams-bootstrap-test"))
+    testImplementation(project(":streams-bootstrap-schema-registry"))
     testImplementation(
         group = "com.bakdata.fluent-kafka-streams-tests",
         name = "schema-registry-mock-junit5",
@@ -42,6 +40,7 @@ dependencies {
     testImplementation(group = "net.mguenther.kafka", name = "kafka-junit", version = kafkaJunitVersion) {
         exclude(group = "org.slf4j", module = "slf4j-log4j12")
     }
+    val confluentVersion: String by project
     testImplementation(group = "io.confluent", name = "kafka-streams-avro-serde", version = confluentVersion)
     val log4jVersion: String by project
     testImplementation(group = "org.apache.logging.log4j", name = "log4j-slf4j2-impl", version = log4jVersion)

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

@@ -24,7 +24,6 @@
 
 package com.bakdata.kafka;
 
-import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -33,28 +32,26 @@
 import org.apache.kafka.streams.StreamsConfig;
 
 /**
- * Configuration to connect to Kafka infrastructure, i.e., bootstrap servers and optionally schema registry.
+ * Configuration to connect to Kafka infrastructure, i.e., bootstrap servers and additional properties such as credentials.
  */
 @Builder
 public class KafkaEndpointConfig {
     private final @NonNull String bootstrapServers;
-    private final String schemaRegistryUrl;
+    @Builder.Default
+    private final Map<String, Object> properties = new HashMap<>();
 
     /**
      * Create Kafka properties to connect to infrastructure.
      * The following properties are configured:
      * <ul>
      *     <li>{@code bootstrap.servers}</li>
-     *     <li>{@code schema.registry.url}</li>
+     *     <li>all properties specified via {@link #properties}</li>
      * </ul>
      * @return properties used for connecting to Kafka
      */
     public Map<String, Object> createKafkaProperties() {
-        final Map<String, String> kafkaConfig = new HashMap<>();
+        final Map<String, Object> kafkaConfig = new HashMap<>(this.properties);
         kafkaConfig.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
-        if (this.schemaRegistryUrl != null) {
-            kafkaConfig.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, this.schemaRegistryUrl);
-        }
         return Collections.unmodifiableMap(kafkaConfig);
     }
 

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

@@ -66,9 +66,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) {
-        SchemaRegistryTopicHook.createSchemaRegistryClient(kafkaProperties)
-                .map(SchemaRegistryTopicHook::new)
-                .ifPresent(configuration::registerTopicHook);
         return new ProducerCleanUpRunner(topics, kafkaProperties, configuration);
     }
 

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

@@ -83,9 +83,6 @@ public static StreamsCleanUpRunner create(final @NonNull Topology topology,
             final @NonNull StreamsConfig streamsConfig, final @NonNull StreamsCleanUpConfiguration configuration) {
         final ImprovedStreamsConfig config = new ImprovedStreamsConfig(streamsConfig);
         final TopologyInformation topologyInformation = new TopologyInformation(topology, config.getAppId());
-        SchemaRegistryTopicHook.createSchemaRegistryClient(config.getKafkaProperties())
-                .map(SchemaRegistryTopicHook::new)
-                .ifPresent(configuration::registerTopicHook);
         return new StreamsCleanUpRunner(topologyInformation, topology, config, configuration);
     }
 

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

@@ -24,14 +24,8 @@
 
 package com.bakdata.kafka.util;
 
-import static com.bakdata.kafka.SchemaRegistryTopicHook.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 ImprovedAdminClient implements AutoCloseable {
 
     private static final Duration ADMIN_TIMEOUT = Duration.ofSeconds(10L);
     private final @NonNull Admin adminClient;
-    private final SchemaRegistryClient schemaRegistryClient;
     private final @NonNull Duration timeout;
 
     /**
@@ -74,10 +67,8 @@ public static ImprovedAdminClient create(@NonNull final Map<String, Object> prop
                     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(timeout)
                 .build();
     }
@@ -86,11 +77,6 @@ public Admin getAdminClient() {
         return new PooledAdmin(this.adminClient);
     }
 
-    public Optional<SchemaRegistryClient> getSchemaRegistryClient() {
-        return Optional.ofNullable(this.schemaRegistryClient)
-                .map(PooledSchemaRegistryClient::new);
-    }
-
     public TopicClient getTopicClient() {
         return new TopicClient(this.getAdminClient(), this.timeout);
     }
@@ -102,13 +88,6 @@ public ConsumerGroupClient getConsumerGroupClient() {
     @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
@@ -126,15 +105,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/test/java/com/bakdata/kafka/ConfiguredProducerAppTest.java

@@ -142,7 +142,7 @@ void shouldThrowIfSchemaRegistryHasBeenConfiguredDifferently() {
                 new ConfiguredProducerApp<>(new TestProducer(), configuration);
         assertThatThrownBy(() -> configuredApp.getKafkaProperties(KafkaEndpointConfig.builder()
                 .bootstrapServers("fake")
-                .schemaRegistryUrl("fake")
+                .properties(Map.of(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "fake"))
                 .build()))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("'schema.registry.url' should not be configured already");

streams-bootstrap-core/src/test/java/com/bakdata/kafka/ConfiguredStreamsAppTest.java

@@ -156,7 +156,7 @@ void shouldThrowIfSchemaRegistryHasBeenConfiguredDifferently() {
                 new ConfiguredStreamsApp<>(new TestApplication(), configuration);
         assertThatThrownBy(() -> configuredApp.getKafkaProperties(KafkaEndpointConfig.builder()
                 .bootstrapServers("fake")
-                .schemaRegistryUrl("fake")
+                .properties(Map.of(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "fake"))
                 .build()))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("'schema.registry.url' should not be configured already");

streams-bootstrap-core/src/test/java/com/bakdata/kafka/integration/KafkaTest.java

@@ -27,6 +27,8 @@
 import com.bakdata.kafka.KafkaEndpointConfig;
 import com.bakdata.kafka.TestUtil;
 import com.bakdata.schemaregistrymock.junit5.SchemaRegistryMockExtension;
+import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
+import java.util.Map;
 import net.mguenther.kafka.junit.EmbeddedKafkaCluster;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -56,7 +58,10 @@ KafkaEndpointConfig createEndpointWithoutSchemaRegistry() {
     KafkaEndpointConfig createEndpoint() {
         return KafkaEndpointConfig.builder()
                 .bootstrapServers(this.kafkaCluster.getBrokerList())
-                .schemaRegistryUrl(this.schemaRegistryMockExtension.getUrl())
+                .properties(Map.of(
+                        AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
+                        this.schemaRegistryMockExtension.getUrl()
+                ))
                 .build();
     }
 }