This is an automated email from the ASF dual-hosted git repository. dannycranmer pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-aws.git
commit 6cacefe76293391b910a4e1212685c502a0b9dc3 Author: Sergey Nuyanzin <[email protected]> AuthorDate: Fri May 26 15:29:12 2023 +0200 [FLINK-32208] Remove dependency on flink-shaded from flink-connector-aws --- .../pom.xml | 10 ++++++++ .../table/test/KinesisFirehoseTableITTest.java | 28 ++++++++++++++++------ .../pom.xml | 18 ++++++++++++++ .../table/test/KinesisStreamsTableApiIT.java | 28 ++++++++++++++++------ .../kinesis/test/KinesisTableApiITCase.java | 27 ++++++++++++++++----- .../flink/streaming/kinesis/test/model/Order.java | 2 +- flink-connector-aws-kinesis-firehose/pom.xml | 15 ++++++++++++ .../sink/testutils/KinesisFirehoseTestUtils.java | 24 +++++++++++++++---- flink-connector-aws-kinesis-streams/pom.xml | 10 ++++++++ .../kinesis/sink/examples/SinkIntoKinesis.java | 22 +++++++++++++---- flink-connector-dynamodb/pom.xml | 6 +++++ .../dynamodb/sink/DynamoDbSinkWriter.java | 3 +-- .../dynamodb/table/DynamoDbDynamicSinkFactory.java | 2 +- .../dynamodb/sink/DynamoDbSinkITCase.java | 3 +-- .../dynamodb/sink/DynamoDbSinkWriterTest.java | 5 ++-- .../dynamodb/sink/DynamoDbWriteRequestTest.java | 3 +-- .../sink/DynamoDbWriterStateSerializerTest.java | 3 +-- .../table/DynamoDbDynamicSinkFactoryTest.java | 3 +-- .../RowDataToAttributeValueConverterTest.java | 3 +-- .../util/DynamoDbSerializationUtilTest.java | 3 +-- .../dynamodb/util/PrimaryKeyBuilderTest.java | 5 ++-- .../kinesis/testutils/KinesisPubsubClient.java | 3 +-- ...inesisDataFetcherForShardConsumerException.java | 2 +- pom.xml | 17 +++---------- tools/maven/checkstyle.xml | 18 +------------- 25 files changed, 178 insertions(+), 85 deletions(-) diff --git a/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-firehose-e2e-tests/pom.xml b/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-firehose-e2e-tests/pom.xml index 8f98b21..5985107 100644 --- a/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-firehose-e2e-tests/pom.xml +++ b/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-firehose-e2e-tests/pom.xml @@ -75,6 +75,16 @@ <artifactId>jackson-databind</artifactId> </dependency> + <dependency> + <groupId>com.fasterxml.jackson.datatype</groupId> + <artifactId>jackson-datatype-jsr310</artifactId> + </dependency> + + <dependency> + <groupId>com.fasterxml.jackson.datatype</groupId> + <artifactId>jackson-datatype-jdk8</artifactId> + </dependency> + <dependency> <groupId>software.amazon.awssdk</groupId> <artifactId>s3</artifactId> diff --git a/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-firehose-e2e-tests/src/test/java/org/apache/flink/connector/firehose/table/test/KinesisFirehoseTableITTest.java b/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-firehose-e2e-tests/src/test/java/org/apache/flink/connector/firehose/table/test/KinesisFirehoseTableITTest.java index 0741ea5..b86c253 100644 --- a/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-firehose-e2e-tests/src/test/java/org/apache/flink/connector/firehose/table/test/KinesisFirehoseTableITTest.java +++ b/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-firehose-e2e-tests/src/test/java/org/apache/flink/connector/firehose/table/test/KinesisFirehoseTableITTest.java @@ -26,13 +26,14 @@ import org.apache.flink.test.resources.ResourceTestUtils; import org.apache.flink.test.util.SQLJobSubmission; import org.apache.flink.util.DockerImageVersions; import org.apache.flink.util.TestLogger; -import org.apache.flink.util.jackson.JacksonMapperFactory; - -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.datatype.jdk8.Jdk8Module; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; import org.assertj.core.api.Assertions; import org.junit.After; import org.junit.AfterClass; @@ -82,7 +83,7 @@ public class KinesisFirehoseTableITTest extends TestLogger { private static final String BUCKET_NAME = "s3-firehose"; private static final String STREAM_NAME = "s3-stream"; - private static final ObjectMapper OBJECT_MAPPER = JacksonMapperFactory.createObjectMapper(); + private static final ObjectMapper OBJECT_MAPPER = createObjectMapper(); private final Path sqlConnectorFirehoseJar = ResourceTestUtils.getResource(".*firehose.jar"); @@ -254,4 +255,17 @@ public class KinesisFirehoseTableITTest extends TestLogger { return "Order{" + "code='" + code + '\'' + ", quantity=" + quantity + '}'; } } + + private static ObjectMapper createObjectMapper() { + ObjectMapper objectMapper = new ObjectMapper(); + registerModules(objectMapper); + return objectMapper; + } + + private static void registerModules(ObjectMapper mapper) { + mapper.registerModule(new JavaTimeModule()) + .registerModule((new Jdk8Module()).configureAbsentsAsNulls(true)) + .disable(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS) + .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + } } diff --git a/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-streams-e2e-tests/pom.xml b/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-streams-e2e-tests/pom.xml index afbe03d..3ce3764 100644 --- a/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-streams-e2e-tests/pom.xml +++ b/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-streams-e2e-tests/pom.xml @@ -69,11 +69,29 @@ </exclusions> </dependency> + <!-- Other third-party dependencies --> + <dependency> + <groupId>com.google.guava</groupId> + <artifactId>guava</artifactId> + </dependency> + <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <scope>test</scope> </dependency> + + <dependency> + <groupId>com.fasterxml.jackson.datatype</groupId> + <artifactId>jackson-datatype-jsr310</artifactId> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>com.fasterxml.jackson.datatype</groupId> + <artifactId>jackson-datatype-jdk8</artifactId> + <scope>test</scope> + </dependency> </dependencies> <build> diff --git a/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-streams-e2e-tests/src/test/java/org/apache/flink/connector/kinesis/table/test/KinesisStreamsTableApiIT.java b/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-streams-e2e-tests/src/test/java/org/apache/flink/connector/kinesis/table/test/KinesisStreamsTableApiIT.java index 6c0a944..312e432 100644 --- a/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-streams-e2e-tests/src/test/java/org/apache/flink/connector/kinesis/table/test/KinesisStreamsTableApiIT.java +++ b/flink-connector-aws-e2e-tests/flink-connector-aws-kinesis-streams-e2e-tests/src/test/java/org/apache/flink/connector/kinesis/table/test/KinesisStreamsTableApiIT.java @@ -27,13 +27,14 @@ import org.apache.flink.connectors.kinesis.testutils.KinesaliteContainer; import org.apache.flink.test.resources.ResourceTestUtils; import org.apache.flink.test.util.SQLJobSubmission; import org.apache.flink.util.DockerImageVersions; -import org.apache.flink.util.jackson.JacksonMapperFactory; - -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.datatype.jdk8.Jdk8Module; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import com.google.common.collect.ImmutableList; import org.assertj.core.api.Assertions; import org.junit.After; import org.junit.AfterClass; @@ -79,7 +80,7 @@ public class KinesisStreamsTableApiIT { private static final String ORDERS_STREAM = "orders"; private static final String INTER_CONTAINER_KINESALITE_ALIAS = "kinesalite"; private static final String DEFAULT_FIRST_SHARD_NAME = "shardId-000000000000"; - private static final ObjectMapper OBJECT_MAPPER = JacksonMapperFactory.createObjectMapper(); + private static final ObjectMapper OBJECT_MAPPER = createObjectMapper(); private SdkHttpClient httpClient; private KinesisClient kinesisClient; @@ -268,4 +269,17 @@ public class KinesisStreamsTableApiIT { return String.format("Order{code: %s, quantity: %d}", code, quantity); } } + + private static ObjectMapper createObjectMapper() { + ObjectMapper objectMapper = new ObjectMapper(); + registerModules(objectMapper); + return objectMapper; + } + + private static void registerModules(ObjectMapper mapper) { + mapper.registerModule(new JavaTimeModule()) + .registerModule((new Jdk8Module()).configureAbsentsAsNulls(true)) + .disable(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS) + .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + } } diff --git a/flink-connector-aws-e2e-tests/flink-connector-kinesis-e2e-tests/src/test/java/org/apache/flink/streaming/kinesis/test/KinesisTableApiITCase.java b/flink-connector-aws-e2e-tests/flink-connector-kinesis-e2e-tests/src/test/java/org/apache/flink/streaming/kinesis/test/KinesisTableApiITCase.java index 5597b57..5642384 100644 --- a/flink-connector-aws-e2e-tests/flink-connector-kinesis-e2e-tests/src/test/java/org/apache/flink/streaming/kinesis/test/KinesisTableApiITCase.java +++ b/flink-connector-aws-e2e-tests/flink-connector-kinesis-e2e-tests/src/test/java/org/apache/flink/streaming/kinesis/test/KinesisTableApiITCase.java @@ -28,12 +28,13 @@ import org.apache.flink.test.resources.ResourceTestUtils; import org.apache.flink.test.util.SQLJobSubmission; import org.apache.flink.util.DockerImageVersions; import org.apache.flink.util.TestLogger; -import org.apache.flink.util.jackson.JacksonMapperFactory; - -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.datatype.jdk8.Jdk8Module; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import com.google.common.collect.ImmutableList; import org.junit.After; import org.junit.AfterClass; import org.junit.Before; @@ -67,7 +68,7 @@ public class KinesisTableApiITCase extends TestLogger { private static final String LARGE_ORDERS_STREAM = "large_orders"; private static final String INTER_CONTAINER_KINESALITE_ALIAS = "kinesalite"; - private static final ObjectMapper OBJECT_MAPPER = JacksonMapperFactory.createObjectMapper(); + private static final ObjectMapper OBJECT_MAPPER = createObjectMapper(); private final Path sqlConnectorKinesisJar = ResourceTestUtils.getResource(".*kinesis.jar"); private static final Network network = Network.newNetwork(); @@ -181,4 +182,18 @@ public class KinesisTableApiITCase extends TestLogger { throw new RuntimeException("Test Failure.", e); } } + + private static com.fasterxml.jackson.databind.ObjectMapper createObjectMapper() { + com.fasterxml.jackson.databind.ObjectMapper objectMapper = + new com.fasterxml.jackson.databind.ObjectMapper(); + registerModules(objectMapper); + return objectMapper; + } + + private static void registerModules(com.fasterxml.jackson.databind.ObjectMapper mapper) { + mapper.registerModule(new JavaTimeModule()) + .registerModule((new Jdk8Module()).configureAbsentsAsNulls(true)) + .disable(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS) + .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + } } diff --git a/flink-connector-aws-e2e-tests/flink-connector-kinesis-e2e-tests/src/test/java/org/apache/flink/streaming/kinesis/test/model/Order.java b/flink-connector-aws-e2e-tests/flink-connector-kinesis-e2e-tests/src/test/java/org/apache/flink/streaming/kinesis/test/model/Order.java index 58cec30..15c158c 100644 --- a/flink-connector-aws-e2e-tests/flink-connector-kinesis-e2e-tests/src/test/java/org/apache/flink/streaming/kinesis/test/model/Order.java +++ b/flink-connector-aws-e2e-tests/flink-connector-kinesis-e2e-tests/src/test/java/org/apache/flink/streaming/kinesis/test/model/Order.java @@ -17,7 +17,7 @@ package org.apache.flink.streaming.kinesis.test.model; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonProperty; import java.util.Objects; diff --git a/flink-connector-aws-kinesis-firehose/pom.xml b/flink-connector-aws-kinesis-firehose/pom.xml index f5106ac..82ad87c 100644 --- a/flink-connector-aws-kinesis-firehose/pom.xml +++ b/flink-connector-aws-kinesis-firehose/pom.xml @@ -126,6 +126,21 @@ under the License. <scope>test</scope> </dependency> + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-core</artifactId> + </dependency> + + <dependency> + <groupId>com.fasterxml.jackson.datatype</groupId> + <artifactId>jackson-datatype-jsr310</artifactId> + </dependency> + + <dependency> + <groupId>com.fasterxml.jackson.datatype</groupId> + <artifactId>jackson-datatype-jdk8</artifactId> + </dependency> + <!-- ArchUit test dependencies --> <dependency> diff --git a/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/testutils/KinesisFirehoseTestUtils.java b/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/testutils/KinesisFirehoseTestUtils.java index f4dee62..0a712fe 100644 --- a/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/testutils/KinesisFirehoseTestUtils.java +++ b/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/testutils/KinesisFirehoseTestUtils.java @@ -20,11 +20,12 @@ package org.apache.flink.connector.firehose.sink.testutils; import org.apache.flink.connector.aws.testutils.AWSServicesTestUtils; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; -import org.apache.flink.util.jackson.JacksonMapperFactory; - -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.datatype.jdk8.Jdk8Module; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; import software.amazon.awssdk.http.SdkHttpClient; import software.amazon.awssdk.services.firehose.FirehoseClient; import software.amazon.awssdk.services.firehose.model.CreateDeliveryStreamRequest; @@ -40,7 +41,7 @@ import java.util.List; */ public class KinesisFirehoseTestUtils { - private static final ObjectMapper MAPPER = JacksonMapperFactory.createObjectMapper(); + private static final ObjectMapper MAPPER = createObjectMapper(); public static FirehoseClient createFirehoseClient(String endpoint, SdkHttpClient httpClient) { return AWSServicesTestUtils.createAwsSyncClient( @@ -83,4 +84,17 @@ public class KinesisFirehoseTestUtils { } return expectedElements; } + + private static ObjectMapper createObjectMapper() { + ObjectMapper objectMapper = new ObjectMapper(); + registerModules(objectMapper); + return objectMapper; + } + + private static void registerModules(ObjectMapper mapper) { + mapper.registerModule(new JavaTimeModule()) + .registerModule((new Jdk8Module()).configureAbsentsAsNulls(true)) + .disable(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS) + .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + } } diff --git a/flink-connector-aws-kinesis-streams/pom.xml b/flink-connector-aws-kinesis-streams/pom.xml index 5e8c454..6a614a6 100644 --- a/flink-connector-aws-kinesis-streams/pom.xml +++ b/flink-connector-aws-kinesis-streams/pom.xml @@ -128,6 +128,16 @@ under the License. <scope>test</scope> </dependency> + <dependency> + <groupId>com.fasterxml.jackson.datatype</groupId> + <artifactId>jackson-datatype-jsr310</artifactId> + </dependency> + + <dependency> + <groupId>com.fasterxml.jackson.datatype</groupId> + <artifactId>jackson-datatype-jdk8</artifactId> + </dependency> + <!-- ArchUit test dependencies --> <dependency> diff --git a/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/examples/SinkIntoKinesis.java b/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/examples/SinkIntoKinesis.java index c861f44..bb7b9a8 100644 --- a/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/examples/SinkIntoKinesis.java +++ b/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/sink/examples/SinkIntoKinesis.java @@ -22,10 +22,11 @@ import org.apache.flink.connector.aws.config.AWSConfigConstants; import org.apache.flink.connector.kinesis.sink.KinesisStreamsSink; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; -import org.apache.flink.util.jackson.JacksonMapperFactory; - -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.datatype.jdk8.Jdk8Module; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; import software.amazon.awssdk.utils.ImmutableMap; @@ -40,7 +41,7 @@ import java.util.Properties; */ public class SinkIntoKinesis { - private static final ObjectMapper OBJECT_MAPPER = JacksonMapperFactory.createObjectMapper(); + private static final ObjectMapper OBJECT_MAPPER = createObjectMapper(); public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); @@ -71,4 +72,17 @@ public class SinkIntoKinesis { env.execute("KDS Async Sink Example Program"); } + + private static ObjectMapper createObjectMapper() { + ObjectMapper objectMapper = new ObjectMapper(); + registerModules(objectMapper); + return objectMapper; + } + + private static void registerModules(ObjectMapper mapper) { + mapper.registerModule(new JavaTimeModule()) + .registerModule((new Jdk8Module()).configureAbsentsAsNulls(true)) + .disable(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS) + .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + } } diff --git a/flink-connector-dynamodb/pom.xml b/flink-connector-dynamodb/pom.xml index 3a4e948..40817d1 100644 --- a/flink-connector-dynamodb/pom.xml +++ b/flink-connector-dynamodb/pom.xml @@ -92,6 +92,12 @@ under the License. <optional>true</optional> </dependency> + <!-- Other third-party dependencies --> + <dependency> + <groupId>com.google.guava</groupId> + <artifactId>guava</artifactId> + </dependency> + <!-- Test dependencies --> <dependency> <groupId>org.apache.flink</groupId> diff --git a/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbSinkWriter.java b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbSinkWriter.java index 2e6b26f..30013b0 100644 --- a/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbSinkWriter.java +++ b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/sink/DynamoDbSinkWriter.java @@ -31,8 +31,7 @@ import org.apache.flink.metrics.Counter; import org.apache.flink.metrics.groups.SinkWriterMetricGroup; import org.apache.flink.util.CollectionUtil; -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap; - +import com.google.common.collect.ImmutableMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; diff --git a/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java index 055caf2..41a7282 100644 --- a/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java +++ b/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactory.java @@ -25,7 +25,7 @@ import org.apache.flink.table.catalog.ResolvedCatalogTable; import org.apache.flink.table.connector.sink.DynamicTableSink; import org.apache.flink.table.factories.FactoryUtil; -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableSet; +import com.google.common.collect.ImmutableSet; import java.util.HashSet; import java.util.Set; diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbSinkITCase.java b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbSinkITCase.java index d73157d..be00a3b 100644 --- a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbSinkITCase.java +++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbSinkITCase.java @@ -32,8 +32,7 @@ import org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator; import org.apache.flink.test.junit5.MiniClusterExtension; import org.apache.flink.util.StringUtils; -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList; - +import com.google.common.collect.ImmutableList; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbSinkWriterTest.java b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbSinkWriterTest.java index 51a500e..60870c5 100644 --- a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbSinkWriterTest.java +++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbSinkWriterTest.java @@ -22,9 +22,8 @@ import org.apache.flink.api.connector.sink2.Sink; import org.apache.flink.connector.base.sink.writer.TestSinkInitContext; import org.apache.flink.connector.dynamodb.sink.client.SdkClientProvider; -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList; -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap; - +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import software.amazon.awssdk.awscore.exception.AwsErrorDetails; import software.amazon.awssdk.core.exception.SdkClientException; diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbWriteRequestTest.java b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbWriteRequestTest.java index f41cac0..779d15c 100644 --- a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbWriteRequestTest.java +++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbWriteRequestTest.java @@ -18,8 +18,7 @@ package org.apache.flink.connector.dynamodb.sink; -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap; - +import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.dynamodb.model.AttributeValue; diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbWriterStateSerializerTest.java b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbWriterStateSerializerTest.java index 3925119..8174817 100644 --- a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbWriterStateSerializerTest.java +++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/sink/DynamoDbWriterStateSerializerTest.java @@ -21,8 +21,7 @@ package org.apache.flink.connector.dynamodb.sink; import org.apache.flink.connector.base.sink.writer.BufferedRequestState; import org.apache.flink.connector.base.sink.writer.ElementConverter; -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap; - +import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.dynamodb.model.AttributeValue; diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactoryTest.java b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactoryTest.java index 97248b8..9a2e276 100644 --- a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactoryTest.java +++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/DynamoDbDynamicSinkFactoryTest.java @@ -31,8 +31,7 @@ import org.apache.flink.table.factories.TableOptionsBuilder; import org.apache.flink.table.factories.TestFormatFactory; import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext; -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap; - +import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import java.util.Collections; diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java index 542addc..1109118 100644 --- a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java +++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/table/RowDataToAttributeValueConverterTest.java @@ -27,8 +27,7 @@ import org.apache.flink.table.data.StringData; import org.apache.flink.table.data.TimestampData; import org.apache.flink.table.types.DataType; -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap; - +import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.dynamodb.model.AttributeValue; diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/util/DynamoDbSerializationUtilTest.java b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/util/DynamoDbSerializationUtilTest.java index 25b3fad..b44d065 100644 --- a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/util/DynamoDbSerializationUtilTest.java +++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/util/DynamoDbSerializationUtilTest.java @@ -21,8 +21,7 @@ package org.apache.flink.connector.dynamodb.util; import org.apache.flink.connector.dynamodb.sink.DynamoDbWriteRequest; import org.apache.flink.connector.dynamodb.sink.DynamoDbWriteRequestType; -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap; - +import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import software.amazon.awssdk.core.SdkBytes; import software.amazon.awssdk.services.dynamodb.model.AttributeValue; diff --git a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/util/PrimaryKeyBuilderTest.java b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/util/PrimaryKeyBuilderTest.java index 663b246..1bf73ba 100644 --- a/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/util/PrimaryKeyBuilderTest.java +++ b/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/util/PrimaryKeyBuilderTest.java @@ -21,9 +21,8 @@ package org.apache.flink.connector.dynamodb.util; import org.apache.flink.connector.dynamodb.sink.InvalidConfigurationException; import org.apache.flink.connector.dynamodb.sink.InvalidRequestException; -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList; -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap; - +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; import software.amazon.awssdk.core.SdkBytes; import software.amazon.awssdk.services.dynamodb.model.AttributeValue; diff --git a/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/KinesisPubsubClient.java b/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/KinesisPubsubClient.java index 992979e..40d1d9f 100644 --- a/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/KinesisPubsubClient.java +++ b/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/KinesisPubsubClient.java @@ -25,8 +25,6 @@ import org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxy; import org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxyInterface; import org.apache.flink.streaming.connectors.kinesis.util.AWSUtil; -import org.apache.flink.shaded.guava30.com.google.common.collect.Lists; - import com.amazonaws.AmazonClientException; import com.amazonaws.auth.AWSCredentialsProvider; import com.amazonaws.client.builder.AwsClientBuilder; @@ -39,6 +37,7 @@ import com.amazonaws.services.kinesis.model.PutRecordsResult; import com.amazonaws.services.kinesis.model.PutRecordsResultEntry; import com.amazonaws.services.kinesis.model.Record; import com.amazonaws.services.kinesis.model.ResourceNotFoundException; +import com.google.common.collect.Lists; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/TestableKinesisDataFetcherForShardConsumerException.java b/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/TestableKinesisDataFetcherForShardConsumerException.java index 2dd91a3..066bb2e 100644 --- a/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/TestableKinesisDataFetcherForShardConsumerException.java +++ b/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/TestableKinesisDataFetcherForShardConsumerException.java @@ -25,7 +25,7 @@ import org.apache.flink.streaming.connectors.kinesis.model.KinesisStreamShardSta import org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxyInterface; import org.apache.flink.streaming.connectors.kinesis.serialization.KinesisDeserializationSchema; -import org.apache.flink.shaded.guava30.com.google.common.util.concurrent.ThreadFactoryBuilder; +import com.google.common.util.concurrent.ThreadFactoryBuilder; import java.util.HashMap; import java.util.LinkedList; diff --git a/pom.xml b/pom.xml index b27c543..a1391b8 100644 --- a/pom.xml +++ b/pom.xml @@ -57,9 +57,9 @@ under the License. <aws.sdkv2.version>2.20.32</aws.sdkv2.version> <netty.version>4.1.86.Final</netty.version> <flink.version>1.16.0</flink.version> - <flink.shaded.version>16.0</flink.shaded.version> - <jackson-bom.version>2.13.4.20221013</jackson-bom.version> + <jackson-bom.version>2.14.3</jackson-bom.version> <glue.schema.registry.version>1.1.14</glue.schema.registry.version> + <guava.version>32.0.0-jre</guava.version> <junit5.version>5.8.1</junit5.version> <assertj.version>3.21.0</assertj.version> @@ -91,11 +91,6 @@ under the License. </modules> <dependencies> - <dependency> - <groupId>org.apache.flink</groupId> - <artifactId>flink-shaded-force-shading</artifactId> - </dependency> - <!-- Test dependencies --> <dependency> <groupId>org.junit.jupiter</groupId> @@ -163,12 +158,6 @@ under the License. <dependencyManagement> <dependencies> - <dependency> - <groupId>org.apache.flink</groupId> - <artifactId>flink-shaded-force-shading</artifactId> - <version>${flink.shaded.version}</version> - </dependency> - <dependency> <groupId>com.amazonaws</groupId> <artifactId>aws-java-sdk-bom</artifactId> @@ -327,7 +316,7 @@ under the License. <dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> - <version>32.0.0-jre</version> + <version>${guava.version}</version> </dependency> <dependency> <groupId>com.amazonaws</groupId> diff --git a/tools/maven/checkstyle.xml b/tools/maven/checkstyle.xml index a46c948..5b90006 100644 --- a/tools/maven/checkstyle.xml +++ b/tools/maven/checkstyle.xml @@ -200,7 +200,7 @@ This file is based on the checkstyle file of Apache Beam. <!-- Checks for out of order import statements. --> <property name="severity" value="error"/> <property name="groups" - value="org.apache.flink,org.apache.flink.shaded,*,javax,java,scala"/> + value="org.apache.flink,*,javax,java,scala"/> <property name="separated" value="true"/> <property name="sortStaticImportsAlphabetically" value="true"/> <property name="option" value="bottom"/> @@ -217,22 +217,6 @@ This file is based on the checkstyle file of Apache Beam. <property name="illegalPkgs" value="autovalue.shaded, avro.shaded, com.google.api.client.repackaged, com.google.appengine.repackaged"/> </module> - <module name="IllegalImport"> - <property name="illegalPkgs" value="org.codehaus.jackson"/> - <message key="import.illegal" value="{0}; Use flink-shaded-jackson instead."/> - </module> - <module name="IllegalImport"> - <property name="illegalPkgs" value="org.objectweb.asm"/> - <message key="import.illegal" value="{0}; Use flink-shaded-asm instead."/> - </module> - <module name="IllegalImport"> - <property name="illegalPkgs" value="io.netty"/> - <message key="import.illegal" value="{0}; Use flink-shaded-netty instead."/> - </module> - <module name="IllegalImport"> - <property name="illegalPkgs" value="com.google.common"/> - <message key="import.illegal" value="{0}; Use flink-shaded-guava instead."/> - </module> <module name="RedundantModifier"> <!-- Checks for redundant modifiers on various symbol definitions.
