Flink 1.18.1 upgrade

.github/workflows/core-build.yml

@@ -0,0 +1,16 @@
+name: PSC-Java Build
+
+on: [pull_request]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v1
+      - name: Set up JDK 1.8
+        uses: actions/setup-java@v1
+        with:
+          java-version: 1.8
+      - name: Build with Maven
+        run: mvn -B install --file pom.xml

.github/workflows/maven.yml → .github/workflows/it-case-tests.yml

@@ -1,4 +1,4 @@
-name: PSC-Java Build
+name: ITCase Tests
 
 on: [pull_request]
 
@@ -13,6 +13,6 @@ jobs:
         with:
           java-version: 1.8
       - name: Build with Maven
-        run: mvn -B install --file pom.xml
+        run: mvn -B install --file pom.xml -DskipTests
       - name: Run psc-flink *ITCase tests
         run: mvn clean test -pl psc-flink -Dtest=*ITCase

psc-flink/pom.xml

@@ -10,8 +10,9 @@
     </parent>
     <artifactId>psc-flink</artifactId>
     <properties>
-        <flink.version>1.15.1</flink.version>
-        <kafka.version>2.8.1</kafka.version>
+        <flink.version>1.18.1</flink.version>
+        <flink.kafka.connector.version>3.2.0-1.18</flink.kafka.connector.version>
+        <kafka.version>3.4.0</kafka.version>
         <zookeeper.version>3.4.10</zookeeper.version>
         <curator.version>2.12.0</curator.version>
         <scala.version>2.12.7</scala.version>
@@ -21,6 +22,19 @@
         <javadoc.opts>-Xdoclint:none</javadoc.opts>
     </properties>
 
+<!--    this is needed due to io.confluent:kafka-schema-registry-client not found in mvn central-->
+    <repositories>
+        <repository>
+            <id>io-confluent</id>
+            <name>io-confluent</name>
+            <url>https://packages.confluent.io/maven</url>
+            <releases>
+                <enabled>true</enabled>
+                <updatePolicy>interval:60</updatePolicy>
+            </releases>
+        </repository>
+    </repositories>
+
     <dependencies>
         <dependency>
             <groupId>org.apache.flink</groupId>
@@ -38,6 +52,18 @@
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-kafka</artifactId>
+            <version>${flink.kafka.connector.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.kafka</groupId>
+                    <artifactId>kafka-clients</artifactId>
+                </exclusion>
+            </exclusions>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-base</artifactId>
             <version>${flink.version}</version>
             <exclusions>
                 <exclusion>
@@ -96,37 +122,6 @@
 
         <!-- 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.12</artifactId>
@@ -176,6 +171,14 @@
             </exclusions>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-base</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+
         <!-- force using the latest zkclient -->
         <dependency>
             <groupId>com.101tec</groupId>
@@ -221,6 +224,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.yaml</groupId>
+            <artifactId>snakeyaml</artifactId>
+            <version>2.2</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-streaming-java</artifactId>

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

@@ -50,7 +50,7 @@ public class PscFlinkConfiguration {
 
     public static TopicUri validateAndGetBaseClusterUri(Properties properties) throws TopicUriSyntaxException {
         if (!properties.containsKey(CLUSTER_URI_CONFIG)) {
-            throw new IllegalArgumentException("Cluster URI is required for transactional producer");
+            throw new IllegalArgumentException("Cluster URI not found");
         }
         return BaseTopicUri.validate(properties.getProperty(CLUSTER_URI_CONFIG));
     }

psc-flink/src/main/java/com/pinterest/flink/connector/psc/dynamic/metadata/ClusterMetadata.java

@@ -0,0 +1,116 @@
+/*
+ * 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.dynamic.metadata;
+
+import com.google.common.base.MoreObjects;
+import com.pinterest.flink.connector.psc.PscFlinkConfiguration;
+import com.pinterest.psc.exception.startup.TopicUriSyntaxException;
+import org.apache.flink.annotation.Experimental;
+
+import java.io.Serializable;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * {@link ClusterMetadata} provides readers information about a cluster on what topics to read and
+ * how to connect to a cluster.
+ */
+@Experimental
+public class ClusterMetadata implements Serializable {
+    private final Set<String> topics;
+    private final Set<String> topicUris;
+    private final Properties properties;
+    private final String clusterUriString;
+
+    /**
+     * Constructs the {@link ClusterMetadata} with the required properties.
+     *
+     * @param topics the topics belonging to a cluster.
+     * @param properties the properties to access a cluster.
+     */
+    public ClusterMetadata(Set<String> topics, Properties properties) {
+        this.topics = topics;
+        this.properties = properties;
+        try {
+            this.clusterUriString = PscFlinkConfiguration.validateAndGetBaseClusterUri(properties).getTopicUriAsString();
+        } catch (TopicUriSyntaxException e) {
+            throw new RuntimeException("Invalid cluster.uri", e);
+        }
+        this.topicUris = topics.stream().map(t -> clusterUriString + t).collect(Collectors.toSet());
+
+    }
+
+    /**
+     * Get the topics.
+     *
+     * @return the topics.
+     */
+    public Set<String> getTopics() {
+        return topics;
+    }
+
+    public Set<String> getTopicUris() {
+        return topicUris;
+    }
+
+    /**
+     * Get the properties.
+     *
+     * @return the properties.
+     */
+    public Properties getProperties() {
+        return properties;
+    }
+
+    /**
+     * Get the cluster URI string.
+     *
+     * @return the cluster URI string.
+     */
+    public String getClusterUriString() {
+        return clusterUriString;
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+                .add("topics", topics)
+                .add("properties", properties)
+                .toString();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ClusterMetadata that = (ClusterMetadata) o;
+        return Objects.equals(topics, that.topics) && Objects.equals(properties, that.properties);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(topics, properties);
+    }
+}

psc-flink/src/main/java/com/pinterest/flink/connector/psc/dynamic/metadata/PscMetadataService.java

@@ -0,0 +1,53 @@
+/*
+ * 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.dynamic.metadata;
+
+import org.apache.flink.annotation.Experimental;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+/** Metadata service that returns PubSub cluster details. */
+@Experimental
+public interface PscMetadataService extends AutoCloseable, Serializable {
+    /**
+     * Get current metadata for all streams.
+     *
+     * @return set of all streams
+     */
+    Set<PscStream> getAllStreams();
+
+    /**
+     * Get current metadata for queried streams.
+     *
+     * @param streamIds stream full names
+     * @return map of stream name to metadata
+     */
+    Map<String, PscStream> describeStreams(Collection<String> streamIds);
+
+    /**
+     * Check if the cluster is active.
+     *
+     * @param clusterId cluster id
+     * @return boolean whether the cluster is active
+     */
+    boolean isClusterActive(String clusterId);
+}