add support for full record (key+value) access (#88)

* add support for full record (key+value) access

- deprecate existing serde methods from interfaces
- add new serde methods tied type-wise to our default impl
- provide default JSON serde implementations
- add tests and utility classes for working with keyed decodable streams
- rewrite existing / add new examples showing how to handle keyless vs keyed records

* add relocation to shadowJar task of example project

this is necessary since decodable flink runtime ships with relocated kafka-clients classes
which otherwise couldn't be resolved on the classpath

the other option would be to relocate these classes in the SDK's build already
but this would still require some modification to this build otherwise tests would
locally fail

* apply same changes to flink 1.18

* apply same changes to flink 1.16

* apply same changes to flink 1.20

* adapt sql to work around limitations and make calcite parser in Flink 1.16 happy

* add necessary relocations for maven build of the example project

* remove some unused imports

* introduce serialization key/value fields constraint checking

- allow for value only constructor in DecodableAbstractStreamRecord
- improve error message and exception handling in Serde code
- adapt usage in job test

* adapt example jobs to reflect recent sdk changes

* apply recent sdk changes for Flink 1.18

* apply recent sdk changes for Flink 1.20

* apply recent sdk changes for Flink 1.16

* minor constructor refactoring

* add test to example project showing how to validate POJO types for serialization

* follow method name convention for parameterized test

* refactor and add building blocks for handling append vs. change streams

* refactor and add tests

* adapt base types for example project

* consistent naming of things in tests

* refactor tests and use consistent namings

* add example job including a test for change stream processing

* add serializability related tests for used POJO types

* clean up example jobs and use more consistent naming

* make sure to serialize record key as JSON only when key Class is set and different from Void

* add example jobs with tests to show how to work with mixed append streams

- keyed2keyless append stream
- keyless2keyed append stream

* add explicit key checks for keyed append and change stream tests of example jobs

* use fixed xmx settings for both builds (maven+gradle)

* add explicit key checks for keyed append and change stream tests in the sdk

* apply latest code / test changes to flink version 1.20

* apply latest code / test changes to flink version 1.18

* apply latest code / test changes to flink version 1.16

Author: hpgrahsl

examples/custom-pipelines-hello-world/build.gradle

@@ -13,13 +13,13 @@ plugins {
 }
 
 group = 'co.decodable.examples'
-version = '0.2'
+version = '0.3'
 
 ext {
     flinkVersion = '1.19.2'
     kafkaConnectorVersion = '3.3.0-1.19'
     log4jVersion = '2.17.1'
-    sdkVersion = '1.19.2-1.0.0.Beta9'
+    sdkVersion = '1.19.2-1.0.0-SNAPSHOT'
 }
 
 repositories {
@@ -116,6 +116,8 @@ shadowJar {
     }
 
     mergeServiceFiles()
+
+    relocate('org.apache.kafka.clients','org.apache.flink.kafka.shaded.org.apache.kafka.clients')
 }
 
 java {
@@ -125,3 +127,7 @@ java {
     withJavadocJar()
     withSourcesJar()
 }
+
+test {
+    maxHeapSize = "1024m"
+}

examples/custom-pipelines-hello-world/decodable-resources.yaml

@@ -123,15 +123,17 @@ spec:
     type: JAVA
 
     # When building with Maven
-    # job_file_path: target/custom-pipelines-hello-world-0.2.jar
+    # job_file_path: target/custom-pipelines-hello-world-0.3.jar
     # When building with Gradle
-    job_file_path: build/libs/custom-pipelines-hello-world-0.2-all.jar
-    
-    # When running the DataStream API Job
-    entry_class: co.decodable.examples.cpdemo.DataStreamJob
-    # When running the Table API Job
-    # entry_class: co.decodable.examples.cpdemo.TableAPIJob
+    job_file_path: build/libs/custom-pipelines-hello-world-0.3-all.jar
 
+    # Select which job to run from the JAR file
+    # NOTE: when running the keyed append stream or change stream based variants of the example jobs,
+    # you'd need to redefine both purchase order stream resource definitions above to reflect this. Also, you'd
+    # need to either feed the source stream from a CDC source or use another pipeline for pre-processing
+    # since the REST resource used in the example currently only supports ingestion into unkeyed append streams.
+    entry_class: co.decodable.examples.cpdemo.KeylessAppendStreamJob
+
     properties:
         flink_version: 1.19-java11
     execution:

examples/custom-pipelines-hello-world/pom.xml

@@ -14,7 +14,7 @@
 
 	<groupId>co.decodable.examples</groupId>
 	<artifactId>custom-pipelines-hello-world</artifactId>
-	<version>0.2</version>
+	<version>0.3</version>
 	<packaging>jar</packaging>
 
 	<name>Decodable Pipeline SDK Example</name>
@@ -25,7 +25,7 @@
 		<scala.binary.version>2.12</scala.binary.version>
 		<maven.compiler.source>${target.java.version}</maven.compiler.source>
 		<maven.compiler.target>${target.java.version}</maven.compiler.target>
-		<decodable.pipeline.sdk.version>1.19.2-1.0.0.Beta9</decodable.pipeline.sdk.version>
+		<decodable.pipeline.sdk.version>1.19.2-1.0.0-SNAPSHOT</decodable.pipeline.sdk.version>
 		<flink.kafka.connector.version>3.3.0-1.19</flink.kafka.connector.version>
 		<aws.msk.iam.auth.version>1.1.6</aws.msk.iam.auth.version>
 		<flink.version>1.19.2</flink.version>
@@ -162,7 +162,7 @@
 			<plugin>
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-shade-plugin</artifactId>
-				<version>3.1.1</version>
+				<version>3.5.3</version>
 				<executions>
 					<!-- Run shade goal on package phase -->
 					<execution>
@@ -179,10 +179,16 @@
 									<exclude>org.xerial.snappy:snappy-java</exclude>
 								</excludes>
 							</artifactSet>
+							<relocations>
+								<relocation>
+									<pattern>org.apache.kafka.clients</pattern>
+									<shadedPattern>org.apache.flink.kafka.shaded.org.apache.kafka.clients</shadedPattern>
+								</relocation>
+							</relocations>
 							<transformers>
 								<transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
 								<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
-									<mainClass>co.decodable.examples.cpdemo.DataStreamJob</mainClass>
+									<mainClass>co.decodable.examples.cpdemo.KeylessAppendStreamJob</mainClass>
 								</transformer>
 							</transformers>
 						</configuration>
@@ -197,6 +203,9 @@
 					<groupId>org.apache.maven.plugins</groupId>
 					<artifactId>maven-surefire-plugin</artifactId>
 					<version>3.1.2</version>
+					<configuration>
+						<argLine>-Xmx1024m</argLine>
+					</configuration>
 				</plugin>
 			</plugins>
 		</pluginManagement>

examples/custom-pipelines-hello-world/src/main/java/co/decodable/examples/cpdemo/ChangeStreamJob.java

@@ -0,0 +1,109 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Copyright Decodable, Inc.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package co.decodable.examples.cpdemo;
+
+import co.decodable.examples.cpdemo.model.OrderKey;
+import co.decodable.examples.cpdemo.model.PurchaseOrder;
+import co.decodable.examples.cpdemo.model.change.KeyedPurchaseOrder;
+import co.decodable.examples.cpdemo.model.change.PurchaseOrderEnvelope;
+import co.decodable.sdk.pipeline.DecodableStreamSink;
+import co.decodable.sdk.pipeline.DecodableStreamSource;
+import co.decodable.sdk.pipeline.metadata.SinkStreams;
+import co.decodable.sdk.pipeline.metadata.SourceStreams;
+import co.decodable.sdk.pipeline.serde.DecodableRecordDeserializationSchema;
+import co.decodable.sdk.pipeline.serde.DecodableRecordSerializationSchema;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.SimpleCounter;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import static co.decodable.examples.cpdemo.ChangeStreamJob.PURCHASE_ORDERS_PROCESSED_STREAM;
+import static co.decodable.examples.cpdemo.ChangeStreamJob.PURCHASE_ORDERS_STREAM;
+
+// spotless:off
+@SourceStreams(PURCHASE_ORDERS_STREAM) // @start region="custom-pipeline"
+@SinkStreams(PURCHASE_ORDERS_PROCESSED_STREAM)
+public class ChangeStreamJob {
+
+  static final String PURCHASE_ORDERS_STREAM = "purchase-orders";
+  static final String PURCHASE_ORDERS_PROCESSED_STREAM = "purchase-orders-processed";
+
+  public static void main(String[] args) throws Exception {
+    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+    // @highlight region regex=".*"
+    DecodableStreamSource<KeyedPurchaseOrder> source = DecodableStreamSource.<KeyedPurchaseOrder>builder()
+        .withStreamName(PURCHASE_ORDERS_STREAM)
+        .withRecordDeserializationSchema(new DecodableRecordDeserializationSchema<>(KeyedPurchaseOrder.class))
+        .build();
+
+    DecodableStreamSink<KeyedPurchaseOrder> sink = DecodableStreamSink.<KeyedPurchaseOrder>builder()
+        .withStreamName(PURCHASE_ORDERS_PROCESSED_STREAM)
+        .withRecordSerializationSchema(new DecodableRecordSerializationSchema<>(OrderKey.class, PurchaseOrder.class))
+        .build();
+    // @end
+
+    DataStream<KeyedPurchaseOrder> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(),
+                    PURCHASE_ORDERS_STREAM)
+        .map(new PurchaseOrderProcessor());
+
+    stream.sinkTo(sink).name(PURCHASE_ORDERS_PROCESSED_STREAM);
+
+    env.execute("purchase order processor with change streams");
+  } // @end region="custom-pipeline"
+
+  public static class PurchaseOrderProcessor extends RichMapFunction<KeyedPurchaseOrder, KeyedPurchaseOrder> {
+
+    private static final long serialVersionUID = 1L;
+    private Counter recordsProcessed;
+
+    // @start region="metric-group"
+    @Override
+    public void open(Configuration parameters) throws Exception {
+      recordsProcessed = getRuntimeContext()
+              .getMetricGroup()
+              .addGroup("DecodableMetrics")
+              .counter("recordsProcessed", new SimpleCounter());
+    }
+    // @end region="metric-group"
+
+    @Override
+    public KeyedPurchaseOrder map(KeyedPurchaseOrder original) throws Exception {
+      var envelope = original.getValue();
+      var processed = new KeyedPurchaseOrder(
+              original.getKey(),
+              new PurchaseOrderEnvelope(
+                      // NOTE:
+                      // This can only work for truly stateless transformations otherwise explicit
+                      // state per key handling is necessary to retrieve the before state accordingly.
+                      processOrder(envelope.getBefore()),
+                      processOrder(envelope.getAfter()),
+                      envelope.getOp(),
+                      envelope.getTs_ms()
+              )
+      );
+      recordsProcessed.inc();
+      return processed;
+    }
+
+    private static PurchaseOrder processOrder(PurchaseOrder original) {
+      return original == null ? null : new PurchaseOrder(
+              original.orderId,
+              original.orderDate,
+              original.customerName.toUpperCase(),
+              original.price,
+              original.productId,
+              original.orderStatus
+      );
+    }
+  }
+}
+//spotless:on

examples/custom-pipelines-hello-world/src/main/java/co/decodable/examples/cpdemo/Keyed2KeylessAppendStreamJob.java

@@ -0,0 +1,93 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Copyright Decodable, Inc.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package co.decodable.examples.cpdemo;
+
+import co.decodable.examples.cpdemo.model.OrderKey;
+import co.decodable.examples.cpdemo.model.PurchaseOrder;
+import co.decodable.examples.cpdemo.model.append.KeyedPurchaseOrder;
+import co.decodable.examples.cpdemo.model.append.KeylessPurchaseOrder;
+import co.decodable.sdk.pipeline.DecodableStreamSink;
+import co.decodable.sdk.pipeline.DecodableStreamSource;
+import co.decodable.sdk.pipeline.metadata.SinkStreams;
+import co.decodable.sdk.pipeline.metadata.SourceStreams;
+import co.decodable.sdk.pipeline.serde.DecodableRecordDeserializationSchema;
+import co.decodable.sdk.pipeline.serde.DecodableRecordSerializationSchema;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.SimpleCounter;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import static co.decodable.examples.cpdemo.KeylessAppendStreamJob.PURCHASE_ORDERS_PROCESSED_STREAM;
+import static co.decodable.examples.cpdemo.KeylessAppendStreamJob.PURCHASE_ORDERS_STREAM;
+
+@SourceStreams(PURCHASE_ORDERS_STREAM)
+@SinkStreams(PURCHASE_ORDERS_PROCESSED_STREAM)
+public class Keyed2KeylessAppendStreamJob {
+
+  static final String PURCHASE_ORDERS_PROCESSED_STREAM = "purchase-orders-processed";
+  static final String PURCHASE_ORDERS_STREAM = "purchase-orders";
+
+  public static void main(String[] args) throws Exception {
+    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+    DecodableStreamSource<KeyedPurchaseOrder> source =
+            DecodableStreamSource.<KeyedPurchaseOrder>builder()
+                    .withStreamName(PURCHASE_ORDERS_STREAM)
+                    .withRecordDeserializationSchema(new DecodableRecordDeserializationSchema<>(KeyedPurchaseOrder.class))
+                    .build();
+
+    DecodableStreamSink<KeylessPurchaseOrder> sink =
+            DecodableStreamSink.<KeylessPurchaseOrder>builder()
+                    .withStreamName(PURCHASE_ORDERS_PROCESSED_STREAM)
+                    .withRecordSerializationSchema(new DecodableRecordSerializationSchema<>(PurchaseOrder.class))
+                    .build();
+
+    DataStream<KeylessPurchaseOrder> stream =
+            env.fromSource(source, WatermarkStrategy.noWatermarks(),
+                            PURCHASE_ORDERS_STREAM)
+                    .map(new NameConverter());
+
+    stream.sinkTo(sink)
+            .name(PURCHASE_ORDERS_PROCESSED_STREAM);
+
+    env.execute("purchase order processor reading keyed/writing keyless append stream");
+  }
+
+  public static class NameConverter extends RichMapFunction<KeyedPurchaseOrder, KeylessPurchaseOrder> {
+
+    private static final long serialVersionUID = 1L;
+
+    private Counter recordsProcessed;
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+      recordsProcessed = getRuntimeContext()
+              .getMetricGroup()
+              .addGroup("DecodableMetrics")
+              .counter("recordsProcessed", new SimpleCounter());
+    }
+
+    @Override
+    public KeylessPurchaseOrder map(KeyedPurchaseOrder order) throws Exception {
+      var orderValue = order.getValue();
+      var newOrder = new PurchaseOrder(
+							orderValue.orderId,
+              orderValue.orderDate,
+              orderValue.customerName.toUpperCase(),
+              orderValue.price,
+              orderValue.productId,
+              orderValue.orderStatus
+      );
+      recordsProcessed.inc();
+      return new KeylessPurchaseOrder(newOrder);
+    }
+  }
+}