Create v3

bakdata · Apr 19, 2024 · 776832c · 776832c
1 parent f09f874
commit 776832c
Show file tree

Hide file tree

Showing 4 changed files with 29 additions and 19 deletions.
diff --git a/streams-bootstrap/src/main/java/com/bakdata/kafka/ConfigurableSerde.java b/streams-bootstrap/src/main/java/com/bakdata/kafka/ConfigurableSerde.java
@@ -31,12 +31,12 @@
 import org.apache.kafka.common.serialization.Serde;
 
 @RequiredArgsConstructor(access = AccessLevel.PACKAGE)
-final class ConfigurableSerde<T> implements Configurable<Serde<T>> {
+final class ConfigurableSerde<S extends Serde<T>, T> implements Configurable<S> {
 
-    private final @NonNull Serde<T> serde;
+    private final @NonNull S serde;
 
     @Override
-    public Serde<T> configure(final Map<String, Object> config, final boolean isKey) {
+    public S configure(final Map<String, Object> config, final boolean isKey) {
         this.serde.configure(config, isKey);
         return this.serde;
     }

diff --git a/streams-bootstrap/src/main/java/com/bakdata/kafka/ConfigurableSerializer.java b/streams-bootstrap/src/main/java/com/bakdata/kafka/ConfigurableSerializer.java
@@ -31,12 +31,12 @@
 import org.apache.kafka.common.serialization.Serializer;
 
 @RequiredArgsConstructor(access = AccessLevel.PACKAGE)
-final class ConfigurableSerializer<T> implements Configurable<Serializer<T>> {
+final class ConfigurableSerializer<S extends Serializer<T>, T> implements Configurable<S> {
 
-    private final @NonNull Serializer<T> serializer;
+    private final @NonNull S serializer;
 
     @Override
-    public Serializer<T> configure(final Map<String, Object> config, final boolean isKey) {
+    public S configure(final Map<String, Object> config, final boolean isKey) {
         this.serializer.configure(config, isKey);
         return this.serializer;
     }

diff --git a/streams-bootstrap/src/main/java/com/bakdata/kafka/Configurator.java b/streams-bootstrap/src/main/java/com/bakdata/kafka/Configurator.java
@@ -129,12 +129,12 @@ public <T> Serializer<T> configureForKeys(final Serializer<T> serializer,
 
     /**
      * Configure a {@code Preconfigured} object using {@link #kafkaProperties}
-     * @param preConfigured pre-configured {@link Serde} or {@link Serializer}
+     * @param preconfigured pre-configured {@link Serde} or {@link Serializer}
      * @return configured instance
      * @param <T> type of configured instance
      */
-    public <T> T configure(final Preconfigured<T> preConfigured) {
-        return preConfigured.configure(this.kafkaProperties);
+    public <T> T configure(final Preconfigured<T> preconfigured) {
+        return preconfigured.configure(this.kafkaProperties);
     }
 
 }
diff --git a/streams-bootstrap/src/main/java/com/bakdata/kafka/Preconfigured.java b/streams-bootstrap/src/main/java/com/bakdata/kafka/Preconfigured.java
@@ -49,9 +49,10 @@ public final class Preconfigured<T> {
      * Pre-configure a {@code Serde} for values
      * @param serde {@code Serde} to pre-configure
      * @return pre-configured serde
+     * @param <S> type of {@link Serde}
      * @param <T> type (de-)serialized by the {@code Serde}
      */
-    public static <T> Preconfigured<Serde<T>> value(final Serde<T> serde) {
+    public static <S extends Serde<T>, T> Preconfigured<S> value(final S serde) {
         return value(configurable(serde));
     }
 
@@ -60,19 +61,22 @@ public static <T> Preconfigured<Serde<T>> value(final Serde<T> serde) {
      * @param serde {@code Serde} to pre-configure
      * @param configOverrides configs passed to {@link Serde#configure(Map, boolean)}
      * @return pre-configured serde
+     * @param <S> type of {@link Serde}
      * @param <T> type (de-)serialized by the {@code Serde}
      */
-    public static <T> Preconfigured<Serde<T>> value(final Serde<T> serde, final Map<String, Object> configOverrides) {
+    public static <S extends Serde<T>, T> Preconfigured<S> value(final S serde,
+            final Map<String, Object> configOverrides) {
         return value(configurable(serde), configOverrides);
     }
 
     /**
      * Pre-configure a {@code Serde} for keys
      * @param serde {@code Serde} to pre-configure
      * @return pre-configured serde
+     * @param <S> type of {@link Serde}
      * @param <T> type (de-)serialized by the {@code Serde}
      */
-    public static <T> Preconfigured<Serde<T>> key(final Serde<T> serde) {
+    public static <S extends Serde<T>, T> Preconfigured<S> key(final S serde) {
         return key(configurable(serde));
     }
 
@@ -81,19 +85,22 @@ public static <T> Preconfigured<Serde<T>> key(final Serde<T> serde) {
      * @param serde {@code Serde} to pre-configure
      * @param configOverrides configs passed to {@link Serde#configure(Map, boolean)}
      * @return pre-configured serde
+     * @param <S> type of {@link Serde}
      * @param <T> type (de-)serialized by the {@code Serde}
      */
-    public static <T> Preconfigured<Serde<T>> key(final Serde<T> serde, final Map<String, Object> configOverrides) {
+    public static <S extends Serde<T>, T> Preconfigured<S> key(final S serde,
+            final Map<String, Object> configOverrides) {
         return key(configurable(serde), configOverrides);
     }
 
     /**
      * Pre-configure a {@code Serializer} for values
      * @param serializer {@code Serializer} to pre-configure
      * @return pre-configured serializer
+     * @param <S> type of {@link Serializer}
      * @param <T> type serialized by the {@code Serializer}
      */
-    public static <T> Preconfigured<Serializer<T>> value(final Serializer<T> serializer) {
+    public static <S extends Serializer<T>, T> Preconfigured<S> value(final S serializer) {
         return value(configurable(serializer));
     }
 
@@ -102,9 +109,10 @@ public static <T> Preconfigured<Serializer<T>> value(final Serializer<T> seriali
      * @param serializer {@code Serializer} to pre-configure
      * @param configOverrides configs passed to {@link Serializer#configure(Map, boolean)}
      * @return pre-configured serializer
+     * @param <S> type of {@link Serializer}
      * @param <T> type serialized by the {@code Serializer}
      */
-    public static <T> Preconfigured<Serializer<T>> value(final Serializer<T> serializer,
+    public static <S extends Serializer<T>, T> Preconfigured<S> value(final S serializer,
             final Map<String, Object> configOverrides) {
         return value(configurable(serializer), configOverrides);
     }
@@ -113,9 +121,10 @@ public static <T> Preconfigured<Serializer<T>> value(final Serializer<T> seriali
      * Pre-configure a {@code Serializer} for keys
      * @param serializer {@code Serializer} to pre-configure
      * @return pre-configured serializer
+     * @param <S> type of {@link Serializer}
      * @param <T> type serialized by the {@code Serializer}
      */
-    public static <T> Preconfigured<Serializer<T>> key(final Serializer<T> serializer) {
+    public static <S extends Serializer<T>, T> Preconfigured<S> key(final S serializer) {
         return key(configurable(serializer));
     }
 
@@ -124,18 +133,19 @@ public static <T> Preconfigured<Serializer<T>> key(final Serializer<T> serialize
      * @param serializer {@code Serializer} to pre-configure
      * @param configOverrides configs passed to {@link Serializer#configure(Map, boolean)}
      * @return pre-configured serializer
+     * @param <S> type of {@link Serializer}
      * @param <T> type serialized by the {@code Serializer}
      */
-    public static <T> Preconfigured<Serializer<T>> key(final Serializer<T> serializer,
+    public static <S extends Serializer<T>, T> Preconfigured<S> key(final S serializer,
             final Map<String, Object> configOverrides) {
         return key(configurable(serializer), configOverrides);
     }
 
-    private static <T> ConfigurableSerde<T> configurable(final Serde<T> serde) {
+    private static <S extends Serde<T>, T> ConfigurableSerde<S, T> configurable(final S serde) {
         return new ConfigurableSerde<>(serde);
     }
 
-    private static <T> ConfigurableSerializer<T> configurable(final Serializer<T> serializer) {
+    private static <S extends Serializer<T>, T> ConfigurableSerializer<S, T> configurable(final S serializer) {
         return new ConfigurableSerializer<>(serializer);
     }