You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "hlteoh37 (via GitHub)" <gi...@apache.org> on 2025/05/28 09:46:57 UTC

[PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

hlteoh37 opened a new pull request, #211:
URL: https://github.com/apache/flink-connector-aws/pull/211

   ## Purpose of the change
   
   Add support for Flink 2.0 for `flink-connector-aws`. This includes:
   1. Kinesis Source/Sink
   2. Firehose Sink
   3. DDB Streams Source
   4. DDB Table Sink
   5. SQS Sink
   
   Refactoring done to ensure compatibility with Flink 2 can be summarised below:
   1. **Flink public Interface changes.** Sink.InitContext has been renamed to WriterInitContext. ParameterTools base package has been migrated.
   **Impact:** Every Sink connector exposes Sink.InitContext as a public interface, because this is exposed in the method signature of the abstract class, AsyncSinkBase. This means a breaking change for the connectors repository to support Flink 2.x
   **Proposal:** I don't see a simple way to decouple the AWS repository code from being tied to either one of Flink 1.x and Flink 2.x. As such I suggest that we split the branches, aws-connector-5.x supports flink-1.20.x and aws-connector-6.x for flink-2.x. This means we have to maintain two active branches (since 1.20 is LTS), but I don't see a better way around this!
   
   2. **Removal of SinkFunction, SourceFunction interfaces.**
   **Impact:** FlinkKinesisConsumer, FlinkKinesisProducer have to be removed.
   **Proposal:** Propose that we remove the above, along with the SQL counterparts, as we have replacements of KinesisStreamsSource and KinesisStreamsSink as replacements.
   **Proposal:** E2E tests and GSR/JSON format tests currently rely on these interfaces, so they will need refactoring as well.
   
   3. **Removal of support for JDK 8, and JDK 17 is now the recommended Java version**
   **Impact/Proposal:** CI/CD can remove testing for JDK 8, and ensure we test against JDK 17
   
   4. **Kryo is disabled by default on Flink 2**
   **Impact/Proposal:** We should sanity check that none of the new states falls back to Kryo. IF there are any, we should make it such that we can use POJOSerializer instead of Kryo.
   
   
   ## Verifying this change
   
   This change is already covered by existing tests, such as existing unit tests, e2e tests.
   
   Will run manual tests for each Source/Sink against local Flink cluster and update here with results.
   
   ## Significant changes
   *(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for convenience.)*
   - [ ] Dependencies have been added or upgraded
   - [ ] Public API has been changed (Public API is any class annotated with `@Public(Evolving)`)
   - [ ] Serializers have been changed
   - [ ] New feature has been introduced
     - If yes, how is this documented? (not applicable / docs / JavaDocs / not documented)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2111480391


##########
pom.xml:
##########
@@ -65,6 +66,7 @@ under the License.
         <assertj.version>3.21.0</assertj.version>
         <archunit.version>0.22.0</archunit.version>
         <testcontainers.version>1.17.2</testcontainers.version>
+        <!--    TODO: Remove Mockito -->

Review Comment:
   +1 to this



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#issuecomment-3005842462

   Thanks for the PR, looks good to me and have approved. 
   
   Ran some quick test to verify it works with Flink 2.0
   
   ### Kinesis Source/Sink
   
   - Set up
   ```
    % ./flink-2.0.0/bin/start-cluster.sh
    % ./flink-2.0.0/bin/sql-client.sh --jar /Users/darenwkt/Documents/github/flink-connector-dynamodb/flink-connector-aws/flink-sql-connector-aws-kinesis-streams/target/flink-sql-connector-aws-kinesis-streams-5.1-SNAPSHOT.jar
   ```
   
   - On SQL client:
     - Create Tables
   ```
   CREATE TABLE random_data (
     `user_id` BIGINT,
     `item_id` BIGINT,
     `category_id` BIGINT,
     `behavior` STRING,
     `ts` TIMESTAMP(3)
   ) WITH (
       'connector' = 'datagen',
       'rows-per-second' = '10'
   );
   
   CREATE TABLE KinesisTable (
     `user_id` BIGINT,
     `item_id` BIGINT,
     `category_id` BIGINT,
     `behavior` STRING,
     `ts` TIMESTAMP(3)
   )
   PARTITIONED BY (user_id, item_id)
   WITH (
     'connector' = 'kinesis',
     'stream.arn' = 'arn:aws:kinesis:us-east-1:redacted:stream/ExampleInputStream',
     'aws.region' = 'us-east-1',
     'source.init.position' = 'LATEST',
     'format' = 'csv'
   );
   ```
   
     - Insert data into SInk
   ```
   INSERT INTO KinesisTable SELECT * FROM random_data;
   ```
     - Read data into Source
   ```
   SELECT * FROM KinesisTable;
   ```
   
     - Verified data is submitted to sink kinesis stream, and read the stream source to verify data is sent
   ![Screenshot 2025-06-25 at 19 24 56](https://github.com/user-attachments/assets/9d1bda27-e566-4001-8bf5-1eaa6b2795df)
   
   
   
   
   ### Firehose Sink
   - Set up
   ```
    % ./flink-2.0.0/bin/start-cluster.sh
    % ./flink-2.0.0/bin/sql-client.sh --jar /Users/darenwkt/Documents/github/flink-connector-dynamodb/flink-connector-aws/flink-sql-connector-aws-kinesis-firehose/target/flink-sql-connector-aws-kinesis-firehose-5.1-SNAPSHOT.jar
   ```
   
   - On SQL client:
     - Create Tables
   ```
   CREATE TABLE random_data (
     `user_id` BIGINT,
     `item_id` BIGINT,
     `category_id` BIGINT,
     `behavior` STRING,
     `ts` TIMESTAMP(3)
   ) WITH (
       'connector' = 'datagen',
       'rows-per-second' = '10'
   );
   
   CREATE TABLE FirehoseTable (
     `user_id` BIGINT,
     `item_id` BIGINT,
     `category_id` BIGINT,
     `behavior` STRING
   )
   WITH (
     'connector' = 'firehose',
     'delivery-stream' = 'user_behavior',
     'aws.region' = 'us-east-2',
     'format' = 'csv'
   );
   ```
   
     - Insert data into SInk
   ```
   INSERT INTO FirehoseTable SELECT * FROM random_data;
   ```
   
     - Verify data appears in Firehose delivery and is found in destination S3: https://analytics-dev-test-redacted.s3.us-east-1.amazonaws.com/2025/06/25/18/analyticsTestOutput-3-2025-06-25-18-33-58-146eb048-00d8-4977-9b7b-e23a17c33963
   
   
   ### DDB Sink
   
   - Set up
   ```
    % ./flink-2.0.0/bin/start-cluster.sh
    % ./flink-2.0.0/bin/sql-client.sh --jar flink-sql-connector-dynamodb-5.1-SNAPSHOT.jar
   ```
   
   - On SQL client:
     - Create Tables
   ```
   CREATE TABLE random_data (
     `user_id` STRING,
     `item_id` BIGINT,
     `category_id` BIGINT,
     `behavior` STRING,
     `ts` TIMESTAMP(3)
   ) WITH (
       'connector' = 'datagen',
       'rows-per-second' = '10'
   );
   
   CREATE TABLE DynamoDbTable (
     `user_id` STRING,
     `item_id` BIGINT,
     `category_id` BIGINT,
     `behavior` STRING,
     `ts` TIMESTAMP(3)
   ) PARTITIONED BY ( user_id )
   WITH (
     'connector' = 'dynamodb',
     'table-name' = 'ddbConnectorTest',
     'aws.region' = 'us-east-1'
   );
   ```
   
     - Insert data into SInk
   ```
   INSERT INTO DynamoDbTable SELECT * FROM random_data;
   ```
   
     - Verify data appears in DDB
   ![Screenshot 2025-06-25 at 19 52 18](https://github.com/user-attachments/assets/605f8f26-b9c2-4f65-987b-195c8d913de1)
   
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#issuecomment-3005430237

   I removed testing for Python because:
   1. It's not succeeding on existing PRs for Flink 1.x
   2. The migration is actually not complete - the real Python artifacts are still vended from the main `apache/flink` repo.
   
   Tracking this improvement work here: https://issues.apache.org/jira/browse/FLINK-38006
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2111469690


##########
pom.xml:
##########
@@ -53,10 +53,11 @@ under the License.
     </scm>
 
     <properties>
+        <!--    TODO: Remove AWS SDK v1 -->
         <aws.sdkv1.version>1.12.439</aws.sdkv1.version>
         <aws.sdkv2.version>2.26.19</aws.sdkv2.version>

Review Comment:
   Suggestion: We can bump sdkv2 version to the latest as well



##########
pom.xml:
##########
@@ -53,10 +53,11 @@ under the License.
     </scm>
 
     <properties>
+        <!--    TODO: Remove AWS SDK v1 -->

Review Comment:
   +1 to this



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2167143818


##########
flink-connector-aws-e2e-tests/flink-formats-avro-glue-schema-registry-e2e-tests/src/test/java/org/apache/flink/glue/schema/registry/test/GSRKinesisPubsubClient.java:
##########
@@ -1,113 +1,113 @@
-/*
- * 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 org.apache.flink.glue.schema.registry.test;
-
-import org.apache.flink.streaming.connectors.kinesis.testutils.KinesisPubsubClient;
-
-import com.amazonaws.services.schemaregistry.common.AWSDeserializerInput;
-import com.amazonaws.services.schemaregistry.common.AWSSerializerInput;
-import com.amazonaws.services.schemaregistry.common.configs.GlueSchemaRegistryConfiguration;
-import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializationFacade;
-import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializationFacade;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import com.amazonaws.services.schemaregistry.utils.AvroRecordType;
-import org.apache.avro.generic.GenericRecord;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.services.glue.model.DataFormat;
-
-import java.nio.ByteBuffer;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-
-/**
- * Simple client to publish and retrieve messages, using the AWS Kinesis SDK, Flink Kinesis
- * Connectors and Glue Schema Registry classes.
- */
-public class GSRKinesisPubsubClient {
-    private final KinesisPubsubClient client;
-    private final GlueSchemaRegistrySerializationFacade serializationFacade;
-    private final GlueSchemaRegistryDeserializationFacade deserializationFacade;
-
-    public GSRKinesisPubsubClient(
-            Properties properties, AwsCredentialsProvider gsrCredentialsProvider) {
-        this.client = new KinesisPubsubClient(properties);
-        this.serializationFacade = createSerializationFacade(gsrCredentialsProvider);
-        this.deserializationFacade = createDeserializationFacade(gsrCredentialsProvider);
-    }
-
-    public void sendMessage(String schema, String streamName, GenericRecord msg) {
-        UUID schemaVersionId =
-                serializationFacade.getOrRegisterSchemaVersion(
-                        AWSSerializerInput.builder()
-                                .schemaDefinition(schema)
-                                .dataFormat(DataFormat.AVRO.name())
-                                .schemaName(streamName)
-                                .transportName(streamName)
-                                .build());
-
-        client.sendMessage(
-                streamName, serializationFacade.serialize(DataFormat.AVRO, msg, schemaVersionId));
-    }
-
-    public List<Object> readAllMessages(String streamName) throws Exception {
-
-        return client.readAllMessages(
-                streamName,
-                bytes ->
-                        deserializationFacade.deserialize(
-                                AWSDeserializerInput.builder()
-                                        .buffer(ByteBuffer.wrap(bytes))
-                                        .transportName(streamName)
-                                        .build()));
-    }
-
-    public void createStream(String stream, int shards, Properties props) throws Exception {
-        client.createTopic(stream, shards, props);
-    }
-
-    private Map<String, Object> getSerDeConfigs() {
-        Map<String, Object> configs = new HashMap<>();
-        configs.put(AWSSchemaRegistryConstants.AWS_REGION, "ca-central-1");
-        configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
-        configs.put(
-                AWSSchemaRegistryConstants.AVRO_RECORD_TYPE,
-                AvroRecordType.GENERIC_RECORD.getName());
-
-        return configs;
-    }
-
-    private GlueSchemaRegistrySerializationFacade createSerializationFacade(
-            final AwsCredentialsProvider credentialsProvider) {
-        return GlueSchemaRegistrySerializationFacade.builder()
-                .credentialProvider(credentialsProvider)
-                .glueSchemaRegistryConfiguration(
-                        new GlueSchemaRegistryConfiguration(getSerDeConfigs()))
-                .build();
-    }
-
-    private GlueSchemaRegistryDeserializationFacade createDeserializationFacade(
-            final AwsCredentialsProvider credentialsProvider) {
-        return GlueSchemaRegistryDeserializationFacade.builder()
-                .credentialProvider(credentialsProvider)
-                .configs(getSerDeConfigs())
-                .build();
-    }
-}
+///*
+// * 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 org.apache.flink.glue.schema.registry.test;
+//
+//import org.apache.flink.streaming.connectors.kinesis.testutils.KinesisPubsubClient;
+//
+//import com.amazonaws.services.schemaregistry.common.AWSDeserializerInput;
+//import com.amazonaws.services.schemaregistry.common.AWSSerializerInput;
+//import com.amazonaws.services.schemaregistry.common.configs.GlueSchemaRegistryConfiguration;
+//import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializationFacade;
+//import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializationFacade;
+//import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
+//import com.amazonaws.services.schemaregistry.utils.AvroRecordType;
+//import org.apache.avro.generic.GenericRecord;
+//import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+//import software.amazon.awssdk.services.glue.model.DataFormat;
+//
+//import java.nio.ByteBuffer;
+//import java.util.HashMap;
+//import java.util.List;
+//import java.util.Map;
+//import java.util.Properties;
+//import java.util.UUID;
+//
+///**
+// * Simple client to publish and retrieve messages, using the AWS Kinesis SDK, Flink Kinesis
+// * Connectors and Glue Schema Registry classes.
+// */
+//public class GSRKinesisPubsubClient {

Review Comment:
   Resolved in PR https://github.com/apache/flink-connector-aws/pull/212



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2135564931


##########
pom.xml:
##########
@@ -53,10 +53,11 @@ under the License.
     </scm>
 
     <properties>
+        <!--    TODO: Remove AWS SDK v1 -->
         <aws.sdkv1.version>1.12.439</aws.sdkv1.version>
         <aws.sdkv2.version>2.26.19</aws.sdkv2.version>

Review Comment:
   Let's decouple the changes -we can always do that in a followup PR



##########
pom.xml:
##########
@@ -53,10 +53,11 @@ under the License.
     </scm>
 
     <properties>
+        <!--    TODO: Remove AWS SDK v1 -->
         <aws.sdkv1.version>1.12.439</aws.sdkv1.version>
         <aws.sdkv2.version>2.26.19</aws.sdkv2.version>

Review Comment:
   Good suggestion, but let's decouple the changes -we can always do that in a followup PR



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "mohsenrezaeithe (via GitHub)" <gi...@apache.org>.
mohsenrezaeithe commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2114711567


##########
.github/workflows/nightly.yml:
##########
@@ -25,8 +25,8 @@ jobs:
     if: github.repository_owner == 'apache'
     strategy:
       matrix:
-        flink: [1.20-SNAPSHOT]
-        java: [ '8, 11, 17']
+        flink: [2.0-SNAPSHOT]
+        java: [ '11, 17']

Review Comment:
   Probably worth adding version 21 to the list just to be inline with Flink's.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2111455330


##########
flink-connector-aws-e2e-tests/flink-formats-avro-glue-schema-registry-e2e-tests/pom.xml:
##########
@@ -53,19 +53,20 @@ under the License.
             <type>test-jar</type>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-kinesis</artifactId>
-            <version>${project.version}</version>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-kinesis</artifactId>
-            <version>${project.version}</version>
-            <type>test-jar</type>
-            <scope>test</scope>
-        </dependency>
+        <!--    TODO: Migrate to flink-connector-aws-kinesis-streams -->

Review Comment:
   +1 on this



##########
flink-connector-aws-e2e-tests/flink-formats-avro-glue-schema-registry-e2e-tests/pom.xml:
##########
@@ -53,19 +53,20 @@ under the License.
             <type>test-jar</type>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-kinesis</artifactId>
-            <version>${project.version}</version>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-kinesis</artifactId>
-            <version>${project.version}</version>
-            <type>test-jar</type>
-            <scope>test</scope>
-        </dependency>
+        <!--    TODO: Migrate to flink-connector-aws-kinesis-streams -->

Review Comment:
   Nice, +1 on this



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#issuecomment-3005300266

   > Quick clarifying question, is the plan to merge this PR and have follow up JIRA to migrate [flink-formats](https://issues.apache.org/jira/browse/FLINK-formats) to use aws-kinesis-streams connector + re-enable test after
   
   We have completed this thanks to your PR! https://github.com/apache/flink-connector-aws/pull/212


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2135560138


##########
flink-connector-aws/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSink.java:
##########
@@ -18,12 +18,13 @@
 package org.apache.flink.connector.firehose.sink;
 
 import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.connector.sink2.StatefulSinkWriter;

Review Comment:
   Hmm, it is used - line 92



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2111467356


##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkITCase.java:
##########
@@ -18,7 +18,9 @@
 
 package org.apache.flink.connector.dynamodb.table;
 
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.commons.io.IOUtils;

Review Comment:
   Worth double checking, don't think this import is used



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2135556413


##########
.github/workflows/nightly.yml:
##########
@@ -25,8 +25,8 @@ jobs:
     if: github.repository_owner == 'apache'
     strategy:
       matrix:
-        flink: [1.20-SNAPSHOT]
-        java: [ '8, 11, 17']
+        flink: [2.0-SNAPSHOT]
+        java: [ '11, 17']

Review Comment:
   Good point - will add



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2135563123


##########
flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkITCase.java:
##########
@@ -18,7 +18,9 @@
 
 package org.apache.flink.connector.dynamodb.table;
 
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.commons.io.IOUtils;

Review Comment:
   It is used - line 133 - the import order was simply rearranged!



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2167064517


##########
pom.xml:
##########
@@ -65,6 +66,7 @@ under the License.
         <assertj.version>3.21.0</assertj.version>
         <archunit.version>0.22.0</archunit.version>
         <testcontainers.version>1.17.2</testcontainers.version>
+        <!--    TODO: Remove Mockito -->

Review Comment:
   Punted this to a followup to simplify this change



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "dieggoluis (via GitHub)" <gi...@apache.org>.
dieggoluis commented on PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#issuecomment-3615812522

   Hi, are there plans to transition the 6.0.0-RC1 to a stable release?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2111462087


##########
flink-connector-aws-e2e-tests/flink-formats-avro-glue-schema-registry-e2e-tests/src/test/java/org/apache/flink/glue/schema/registry/test/GSRKinesisPubsubClient.java:
##########
@@ -1,113 +1,113 @@
-/*
- * 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 org.apache.flink.glue.schema.registry.test;
-
-import org.apache.flink.streaming.connectors.kinesis.testutils.KinesisPubsubClient;
-
-import com.amazonaws.services.schemaregistry.common.AWSDeserializerInput;
-import com.amazonaws.services.schemaregistry.common.AWSSerializerInput;
-import com.amazonaws.services.schemaregistry.common.configs.GlueSchemaRegistryConfiguration;
-import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializationFacade;
-import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializationFacade;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import com.amazonaws.services.schemaregistry.utils.AvroRecordType;
-import org.apache.avro.generic.GenericRecord;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.services.glue.model.DataFormat;
-
-import java.nio.ByteBuffer;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-
-/**
- * Simple client to publish and retrieve messages, using the AWS Kinesis SDK, Flink Kinesis
- * Connectors and Glue Schema Registry classes.
- */
-public class GSRKinesisPubsubClient {
-    private final KinesisPubsubClient client;
-    private final GlueSchemaRegistrySerializationFacade serializationFacade;
-    private final GlueSchemaRegistryDeserializationFacade deserializationFacade;
-
-    public GSRKinesisPubsubClient(
-            Properties properties, AwsCredentialsProvider gsrCredentialsProvider) {
-        this.client = new KinesisPubsubClient(properties);
-        this.serializationFacade = createSerializationFacade(gsrCredentialsProvider);
-        this.deserializationFacade = createDeserializationFacade(gsrCredentialsProvider);
-    }
-
-    public void sendMessage(String schema, String streamName, GenericRecord msg) {
-        UUID schemaVersionId =
-                serializationFacade.getOrRegisterSchemaVersion(
-                        AWSSerializerInput.builder()
-                                .schemaDefinition(schema)
-                                .dataFormat(DataFormat.AVRO.name())
-                                .schemaName(streamName)
-                                .transportName(streamName)
-                                .build());
-
-        client.sendMessage(
-                streamName, serializationFacade.serialize(DataFormat.AVRO, msg, schemaVersionId));
-    }
-
-    public List<Object> readAllMessages(String streamName) throws Exception {
-
-        return client.readAllMessages(
-                streamName,
-                bytes ->
-                        deserializationFacade.deserialize(
-                                AWSDeserializerInput.builder()
-                                        .buffer(ByteBuffer.wrap(bytes))
-                                        .transportName(streamName)
-                                        .build()));
-    }
-
-    public void createStream(String stream, int shards, Properties props) throws Exception {
-        client.createTopic(stream, shards, props);
-    }
-
-    private Map<String, Object> getSerDeConfigs() {
-        Map<String, Object> configs = new HashMap<>();
-        configs.put(AWSSchemaRegistryConstants.AWS_REGION, "ca-central-1");
-        configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
-        configs.put(
-                AWSSchemaRegistryConstants.AVRO_RECORD_TYPE,
-                AvroRecordType.GENERIC_RECORD.getName());
-
-        return configs;
-    }
-
-    private GlueSchemaRegistrySerializationFacade createSerializationFacade(
-            final AwsCredentialsProvider credentialsProvider) {
-        return GlueSchemaRegistrySerializationFacade.builder()
-                .credentialProvider(credentialsProvider)
-                .glueSchemaRegistryConfiguration(
-                        new GlueSchemaRegistryConfiguration(getSerDeConfigs()))
-                .build();
-    }
-
-    private GlueSchemaRegistryDeserializationFacade createDeserializationFacade(
-            final AwsCredentialsProvider credentialsProvider) {
-        return GlueSchemaRegistryDeserializationFacade.builder()
-                .credentialProvider(credentialsProvider)
-                .configs(getSerDeConfigs())
-                .build();
-    }
-}
+///*
+// * 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 org.apache.flink.glue.schema.registry.test;
+//
+//import org.apache.flink.streaming.connectors.kinesis.testutils.KinesisPubsubClient;
+//
+//import com.amazonaws.services.schemaregistry.common.AWSDeserializerInput;
+//import com.amazonaws.services.schemaregistry.common.AWSSerializerInput;
+//import com.amazonaws.services.schemaregistry.common.configs.GlueSchemaRegistryConfiguration;
+//import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializationFacade;
+//import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializationFacade;
+//import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
+//import com.amazonaws.services.schemaregistry.utils.AvroRecordType;
+//import org.apache.avro.generic.GenericRecord;
+//import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+//import software.amazon.awssdk.services.glue.model.DataFormat;
+//
+//import java.nio.ByteBuffer;
+//import java.util.HashMap;
+//import java.util.List;
+//import java.util.Map;
+//import java.util.Properties;
+//import java.util.UUID;
+//
+///**
+// * Simple client to publish and retrieve messages, using the AWS Kinesis SDK, Flink Kinesis
+// * Connectors and Glue Schema Registry classes.
+// */
+//public class GSRKinesisPubsubClient {

Review Comment:
   I think instead of disabling the e2e test in this PR, we could consider:
   
   1. Migrate to flink-connector-aws-kinesis-streams as part of this PR so that we have these tests working?
   2. OR Migrate before this PR as a separate story and pre-requisite to move to Flink 2.0?
   
   Any thoughts?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "darenwkt (via GitHub)" <gi...@apache.org>.
darenwkt commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2111464018


##########
flink-connector-aws/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSink.java:
##########
@@ -18,12 +18,13 @@
 package org.apache.flink.connector.firehose.sink;
 
 import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.connector.sink2.StatefulSinkWriter;

Review Comment:
   Worth double checking, don't think this import is used



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2135560138


##########
flink-connector-aws/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSink.java:
##########
@@ -18,12 +18,13 @@
 package org.apache.flink.connector.firehose.sink;
 
 import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.connector.sink2.StatefulSinkWriter;

Review Comment:
   Hmm, it is used - line 96



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-37687] Introduce support for Flink 2.0 in flink-connector-aws [flink-connector-aws]

Posted by "hlteoh37 (via GitHub)" <gi...@apache.org>.
hlteoh37 commented on code in PR #211:
URL: https://github.com/apache/flink-connector-aws/pull/211#discussion_r2135558482


##########
flink-connector-aws-e2e-tests/flink-formats-avro-glue-schema-registry-e2e-tests/pom.xml:
##########
@@ -53,19 +53,20 @@ under the License.
             <type>test-jar</type>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-kinesis</artifactId>
-            <version>${project.version}</version>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-kinesis</artifactId>
-            <version>${project.version}</version>
-            <type>test-jar</type>
-            <scope>test</scope>
-        </dependency>
+        <!--    TODO: Migrate to flink-connector-aws-kinesis-streams -->

Review Comment:
   https://github.com/apache/flink-connector-aws/pull/212



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org