Implement PscSource, PscSink, and Table API's to close the gap on Flink 1.15.1 API's

.github/workflows/maven.yml

@@ -13,4 +13,6 @@ jobs:
         with:
           java-version: 1.8
       - name: Build with Maven
-        run: mvn -B package --file pom.xml
+        run: mvn -B install --file pom.xml
+      - name: Run psc-flink *ITCase tests
+        run: mvn clean test -pl psc-flink -Dtest=*ITCase

.gitignore

@@ -22,3 +22,6 @@ buildNumber.properties
 
 # vscode
 .vscode
+
+#test files
+psc-flink/orgapacheflinkutilNetUtils*

pom.xml

@@ -23,6 +23,8 @@
         <gson.version>2.8.6</gson.version>
         <guava.version>29.0-jre</guava.version>
         <junit.jupiter.version>5.6.2</junit.jupiter.version>
+        <slf4j.version>1.7.32</slf4j.version>
+        <log4j.version>1.2.17</log4j.version>
         <log4j2.version>2.17.1</log4j2.version>
         <mockito.version>3.4.6</mockito.version>
         <reflections.version>0.9.9</reflections.version>

psc-examples/src/main/java/com/pinterest/psc/example/kafka/ExamplePscProducer.java

@@ -9,12 +9,14 @@
 import com.pinterest.psc.serde.IntegerSerializer;
 import com.pinterest.psc.serde.StringSerializer;
 
+import java.io.IOException;
+
 public class ExamplePscProducer {
 
     private static final PscLogger logger = PscLogger.getLogger(ExamplePscProducer.class);
     private static final int NUM_MESSAGES = 10;
 
-    public static void main(String[] args) throws ConfigurationException, ProducerException {
+    public static void main(String[] args) throws ConfigurationException, ProducerException, IOException {
         if (args.length < 1) {
             logger.error("ExamplePscProducer needs one argument: topicUri");
             return;

psc-examples/src/main/java/com/pinterest/psc/example/migration/producer/Psc.java

@@ -1,5 +1,6 @@
 package com.pinterest.psc.example.migration.producer;
 
+import java.io.IOException;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import com.pinterest.psc.config.PscConfiguration;
@@ -11,7 +12,7 @@
 
 public class Psc {
 
-    static public void main(String[] args) throws ConfigurationException, ProducerException {
+    static public void main(String[] args) throws ConfigurationException, ProducerException, IOException {
         String topicUri = "plaintext:/rn:kafka:dev:local-cloud_local-region::local-cluster:my_test_topic";
 
         PscConfiguration pscConfiguration = new PscConfiguration();

psc-flink/pom.xml

@@ -11,11 +11,10 @@
     <artifactId>psc-flink</artifactId>
     <properties>
         <flink.version>1.15.1</flink.version>
-        <flink.old.version>1.11.1</flink.old.version>
         <kafka.version>2.8.1</kafka.version>
         <zookeeper.version>3.4.10</zookeeper.version>
         <curator.version>2.12.0</curator.version>
-        <scala.version>2.11.12</scala.version>
+        <scala.version>2.12.7</scala.version>
         <scala.binary.version>2.12</scala.binary.version>
         <powermock.version>2.0.0-RC.4</powermock.version>
         <hamcrest.version>1.3</hamcrest.version>
@@ -41,10 +40,6 @@
             <artifactId>flink-connector-kafka</artifactId>
             <version>${flink.version}</version>
             <exclusions>
-                <exclusion>
-                    <groupId>org.apache.flink</groupId>
-                    <artifactId>force-shading</artifactId>
-                </exclusion>
                 <exclusion>
                     <groupId>org.apache.kafka</groupId>
                     <artifactId>kafka-clients</artifactId>
@@ -53,19 +48,12 @@
             <scope>provided</scope>
         </dependency>
 
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>force-shading</artifactId>
-            <version>${flink.old.version}</version>
-            <scope>provided</scope>
-        </dependency>
-
         <!-- Table ecosystem -->
         <!-- Projects depending on this project won't depend on flink-table-*. -->
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
-            <version>${flink.old.version}</version>
+            <artifactId>flink-table-api-java-bridge</artifactId>
+            <version>${flink.version}</version>
             <scope>provided</scope>
             <optional>true</optional>
             <exclusions>
@@ -88,20 +76,6 @@
             <optional>true</optional>
         </dependency>
 
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
-            <version>${flink.old.version}</version>
-            <scope>provided</scope>
-            <optional>true</optional>
-            <exclusions>
-                <exclusion>
-                    <groupId>org.apache.flink</groupId>
-                    <artifactId>flink-table-common</artifactId>
-                </exclusion>
-            </exclusions>
-        </dependency>
-
         <dependency>
             <groupId>com.pinterest.psc</groupId>
             <artifactId>psc</artifactId>
@@ -122,9 +96,40 @@
 
         <!-- test dependencies -->
 
+        <dependency>
+            <groupId>com.pinterest.psc</groupId>
+            <artifactId>psc-logging</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+            <version>${slf4j.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>log4j</groupId>
+            <artifactId>log4j</artifactId>
+            <version>${log4j.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>log4j</groupId>
+            <artifactId>apache-log4j-extras</artifactId>
+            <version>${log4j.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>net.logstash.log4j</groupId>
+            <artifactId>jsonevent-layout</artifactId>
+            <version>1.7</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka_2.13</artifactId>
+            <artifactId>kafka_2.12</artifactId>
             <version>${kafka.version}</version>
             <scope>test</scope>
             <exclusions>
@@ -239,8 +244,15 @@
 
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-test-utils_2.11</artifactId>
-            <version>${flink.old.version}</version>
+            <artifactId>flink-connector-test-utils</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-test-utils</artifactId>
+            <version>${flink.version}</version>
             <scope>test</scope>
             <exclusions>
                 <exclusion>
@@ -304,29 +316,23 @@
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
-            <version>${flink.old.version}</version>
-            <type>test-jar</type>
+            <version>${flink.version}</version>
             <scope>test</scope>
-            <exclusions>
-                <exclusion>
-                    <groupId>org.apache.flink</groupId>
-                    <artifactId>flink-table-common</artifactId>
-                </exclusion>
-            </exclusions>
         </dependency>
 
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
-            <version>${flink.old.version}</version>
+            <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
             <type>test-jar</type>
             <scope>test</scope>
-            <exclusions>
-                <exclusion>
-                    <groupId>org.apache.flink</groupId>
-                    <artifactId>flink-table-common</artifactId>
-                </exclusion>
-            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
         </dependency>
 
         <dependency>
@@ -373,6 +379,20 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>kafka</artifactId>
+            <version>1.16.2</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>1.16.2</version>
+            <scope>test</scope>
+        </dependency>
+
         <!-- Kafka table descriptor testing -->
         <dependency>
             <groupId>org.apache.flink</groupId>
@@ -382,6 +402,32 @@
             <scope>test</scope>
         </dependency>
 
+        <!-- Kafka SQL IT test with formats -->
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-json</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-avro</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-avro-confluent-registry</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-csv</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
 
     <build>

psc-flink/src/main/java/com/pinterest/flink/connector/psc/MetricUtil.java

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.pinterest.flink.connector.psc;
+
+import com.pinterest.psc.metrics.Metric;
+import com.pinterest.psc.metrics.MetricName;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Counter;
+
+import java.util.Map;
+import java.util.function.Predicate;
+
+/** Collection of methods to interact with PSC's client metric system. */
+@Internal
+public class MetricUtil {
+
+    /**
+     * Tries to find the PSC {@link Metric} in the provided metrics.
+     *
+     * @return {@link Metric} which exposes continuous updates
+     * @throws IllegalStateException if the metric is not part of the provided metrics
+     */
+    public static Metric getPscMetric(
+            Map<MetricName, ? extends Metric> metrics, String metricGroup, String metricName) {
+        return getPscMetric(
+                metrics,
+                e ->
+                        e.getKey().group().equals(metricGroup)
+                                && e.getKey().name().equals(metricName));
+    }
+
+    /**
+     * Tries to find the PSC {@link Metric} in the provided metrics matching a given filter.
+     *
+     * @return {@link Metric} which exposes continuous updates
+     * @throws IllegalStateException if no metric matches the given filter
+     */
+    public static Metric getPscMetric(
+            Map<MetricName, ? extends Metric> metrics,
+            Predicate<Map.Entry<MetricName, ? extends Metric>> filter) {
+        return metrics.entrySet().stream()
+                .filter(filter)
+                .map(Map.Entry::getValue)
+                .findFirst()
+                .orElseThrow(
+                        () ->
+                                new IllegalStateException(
+                                        "Cannot find PSC metric matching current filter."));
+    }
+
+    /**
+     * Ensures that the counter has the same value as the given Psc metric.
+     *
+     * <p>Do not use this method for every record because {@link Metric#metricValue()} is an
+     * expensive operation.
+     *
+     * @param from PSC's {@link Metric} to query
+     * @param to {@link Counter} to write the value to
+     */
+    public static void sync(Metric from, Counter to) {
+        to.inc(((Number) from.metricValue()).longValue() - to.getCount());
+    }
+}