You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ho...@apache.org on 2023/09/04 15:37:30 UTC
[flink-connector-aws] 02/02: [hotfix] Remove dependency on shaded guava to reduce refactoring needed for Flink version upgrades
This is an automated email from the ASF dual-hosted git repository.
hong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-aws.git
commit 8b0ae0f45fea40beb52e12e6a25ede6003bef1be
Author: Hong Liang Teoh <li...@amazon.com>
AuthorDate: Mon Aug 28 17:43:08 2023 +0100
[hotfix] Remove dependency on shaded guava to reduce refactoring needed for Flink version upgrades
---
.../enumerator/KinesisStreamsSourceEnumerator.java | 22 ++--
...StreamsSourceEnumeratorStateSerializerTest.java | 32 +++---
.../KinesisStreamsSourceEnumeratorTest.java | 11 +-
.../source/proxy/KinesisStreamProxyTest.java | 53 +++++-----
.../reader/KinesisStreamsRecordEmitterTest.java | 113 ++++++++++++---------
.../reader/PollingKinesisShardSplitReaderTest.java | 38 +++----
6 files changed, 146 insertions(+), 123 deletions(-)
diff --git a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/enumerator/KinesisStreamsSourceEnumerator.java b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/enumerator/KinesisStreamsSourceEnumerator.java
index 14bc65d..9f8b7f0 100644
--- a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/enumerator/KinesisStreamsSourceEnumerator.java
+++ b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/enumerator/KinesisStreamsSourceEnumerator.java
@@ -30,10 +30,6 @@ import org.apache.flink.connector.kinesis.source.proxy.StreamProxy;
import org.apache.flink.connector.kinesis.source.split.KinesisShardSplit;
import org.apache.flink.connector.kinesis.source.split.StartingPosition;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableSet;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.services.kinesis.model.Shard;
@@ -303,24 +299,26 @@ public class KinesisStreamsSourceEnumerator
private ShardAssignerContext withPendingSplitAssignments(
Map<Integer, List<KinesisShardSplit>> pendingSplitAssignments) {
- ImmutableMap.Builder<Integer, List<KinesisShardSplit>> mapBuilder =
- ImmutableMap.builder();
+ Map<Integer, List<KinesisShardSplit>> copyPendingSplitAssignments = new HashMap<>();
for (Entry<Integer, List<KinesisShardSplit>> entry :
pendingSplitAssignments.entrySet()) {
- mapBuilder.put(entry.getKey(), ImmutableList.copyOf(entry.getValue()));
+ copyPendingSplitAssignments.put(
+ entry.getKey(),
+ Collections.unmodifiableList(new ArrayList<>(entry.getValue())));
}
- this.pendingSplitAssignments = mapBuilder.build();
+ this.pendingSplitAssignments = Collections.unmodifiableMap(copyPendingSplitAssignments);
return this;
}
@Override
public Map<Integer, Set<KinesisShardSplit>> getCurrentSplitAssignment() {
- ImmutableMap.Builder<Integer, Set<KinesisShardSplit>> mapBuilder =
- ImmutableMap.builder();
+ Map<Integer, Set<KinesisShardSplit>> copyCurrentSplitAssignment = new HashMap<>();
for (Entry<Integer, Set<KinesisShardSplit>> entry : splitAssignment.entrySet()) {
- mapBuilder.put(entry.getKey(), ImmutableSet.copyOf(entry.getValue()));
+ copyCurrentSplitAssignment.put(
+ entry.getKey(),
+ Collections.unmodifiableSet(new HashSet<>(entry.getValue())));
}
- return mapBuilder.build();
+ return Collections.unmodifiableMap(copyCurrentSplitAssignment);
}
@Override
diff --git a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/enumerator/KinesisStreamsSourceEnumeratorStateSerializerTest.java b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/enumerator/KinesisStreamsSourceEnumeratorStateSerializerTest.java
index 4a01f68..79b9e5a 100644
--- a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/enumerator/KinesisStreamsSourceEnumeratorStateSerializerTest.java
+++ b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/enumerator/KinesisStreamsSourceEnumeratorStateSerializerTest.java
@@ -22,12 +22,12 @@ import org.apache.flink.connector.kinesis.source.split.KinesisShardSplit;
import org.apache.flink.connector.kinesis.source.split.KinesisShardSplitSerializer;
import org.apache.flink.core.io.VersionMismatchException;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableSet;
-
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static java.util.Arrays.copyOf;
import static org.apache.flink.connector.kinesis.source.util.TestUtil.generateShardId;
@@ -40,10 +40,11 @@ class KinesisStreamsSourceEnumeratorStateSerializerTest {
@Test
void testSerializeAndDeserializeEverythingSpecified() throws Exception {
Set<String> completedShardIds =
- ImmutableSet.of(
- "shardId-000000000001", "shardId-000000000002", "shardId-000000000003");
+ Stream.of("shardId-000000000001", "shardId-000000000002", "shardId-000000000003")
+ .collect(Collectors.toSet());
Set<KinesisShardSplit> unassignedSplits =
- ImmutableSet.of(getTestSplit(generateShardId(1)), getTestSplit(generateShardId(2)));
+ Stream.of(getTestSplit(generateShardId(1)), getTestSplit(generateShardId(2)))
+ .collect(Collectors.toSet());
String lastSeenShardId = "shardId-000000000002";
KinesisStreamsSourceEnumeratorState initialState =
new KinesisStreamsSourceEnumeratorState(unassignedSplits, lastSeenShardId);
@@ -62,10 +63,11 @@ class KinesisStreamsSourceEnumeratorStateSerializerTest {
@Test
void testDeserializeWithWrongVersionStateSerializer() throws Exception {
Set<String> completedShardIds =
- ImmutableSet.of(
- "shardId-000000000001", "shardId-000000000002", "shardId-000000000003");
+ Stream.of("shardId-000000000001", "shardId-000000000002", "shardId-000000000003")
+ .collect(Collectors.toSet());
Set<KinesisShardSplit> unassignedSplits =
- ImmutableSet.of(getTestSplit(generateShardId(1)), getTestSplit(generateShardId(2)));
+ Stream.of(getTestSplit(generateShardId(1)), getTestSplit(generateShardId(2)))
+ .collect(Collectors.toSet());
String lastSeenShardId = "shardId-000000000002";
KinesisStreamsSourceEnumeratorState initialState =
new KinesisStreamsSourceEnumeratorState(unassignedSplits, lastSeenShardId);
@@ -92,10 +94,11 @@ class KinesisStreamsSourceEnumeratorStateSerializerTest {
@Test
void testDeserializeWithWrongVersionSplitSerializer() throws Exception {
Set<String> completedShardIds =
- ImmutableSet.of(
- "shardId-000000000001", "shardId-000000000002", "shardId-000000000003");
+ Stream.of("shardId-000000000001", "shardId-000000000002", "shardId-000000000003")
+ .collect(Collectors.toSet());
Set<KinesisShardSplit> unassignedSplits =
- ImmutableSet.of(getTestSplit(generateShardId(1)), getTestSplit(generateShardId(2)));
+ Stream.of(getTestSplit(generateShardId(1)), getTestSplit(generateShardId(2)))
+ .collect(Collectors.toSet());
String lastSeenShardId = "shardId-000000000002";
KinesisStreamsSourceEnumeratorState initialState =
new KinesisStreamsSourceEnumeratorState(unassignedSplits, lastSeenShardId);
@@ -120,10 +123,11 @@ class KinesisStreamsSourceEnumeratorStateSerializerTest {
@Test
void testSerializeWithTrailingBytes() throws Exception {
Set<String> completedShardIds =
- ImmutableSet.of(
- "shardId-000000000001", "shardId-000000000002", "shardId-000000000003");
+ Stream.of("shardId-000000000001", "shardId-000000000002", "shardId-000000000003")
+ .collect(Collectors.toSet());
Set<KinesisShardSplit> unassignedSplits =
- ImmutableSet.of(getTestSplit(generateShardId(1)), getTestSplit(generateShardId(2)));
+ Stream.of(getTestSplit(generateShardId(1)), getTestSplit(generateShardId(2)))
+ .collect(Collectors.toSet());
String lastSeenShardId = "shardId-000000000002";
KinesisStreamsSourceEnumeratorState initialState =
new KinesisStreamsSourceEnumeratorState(unassignedSplits, lastSeenShardId);
diff --git a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/enumerator/KinesisStreamsSourceEnumeratorTest.java b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/enumerator/KinesisStreamsSourceEnumeratorTest.java
index da9d344..b6056e4 100644
--- a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/enumerator/KinesisStreamsSourceEnumeratorTest.java
+++ b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/enumerator/KinesisStreamsSourceEnumeratorTest.java
@@ -29,9 +29,6 @@ import org.apache.flink.connector.kinesis.source.split.KinesisShardSplit;
import org.apache.flink.connector.kinesis.source.util.TestUtil;
import org.apache.flink.util.FlinkRuntimeException;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableSet;
-
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
@@ -163,7 +160,7 @@ class KinesisStreamsSourceEnumeratorTest {
TestKinesisStreamProxy streamProxy = getTestStreamProxy();
final String completedShard = generateShardId(0);
final String lastSeenShard = generateShardId(1);
- final Set<String> completedShardIds = ImmutableSet.of(completedShard);
+ final Set<String> completedShardIds = Collections.singleton(completedShard);
KinesisStreamsSourceEnumeratorState state =
new KinesisStreamsSourceEnumeratorState(Collections.emptySet(), lastSeenShard);
@@ -253,7 +250,7 @@ class KinesisStreamsSourceEnumeratorTest {
// Given one shard split is returned
KinesisShardSplit returnedSplit =
initialSplitAssignment.assignment().get(subtaskId).get(0);
- enumerator.addSplitsBack(ImmutableList.of(returnedSplit), subtaskId);
+ enumerator.addSplitsBack(Collections.singletonList(returnedSplit), subtaskId);
// When first periodic discovery runs
context.runPeriodicCallable(0);
@@ -273,7 +270,7 @@ class KinesisStreamsSourceEnumeratorTest {
TestKinesisStreamProxy streamProxy = getTestStreamProxy();
KinesisStreamsSourceEnumerator enumerator =
getSimpleEnumeratorWithNoState(context, streamProxy);
- List<KinesisShardSplit> splits = ImmutableList.of(getTestSplit());
+ List<KinesisShardSplit> splits = Collections.singletonList(getTestSplit());
// Given enumerator has no assigned splits
// When we add splits back
@@ -316,7 +313,7 @@ class KinesisStreamsSourceEnumeratorTest {
// When we add splits back
KinesisShardSplit returnedSplit =
initialSplitAssignment.assignment().get(subtaskId).get(0);
- enumerator.addSplitsBack(ImmutableList.of(returnedSplit), 1);
+ enumerator.addSplitsBack(Collections.singletonList(returnedSplit), 1);
// Then splits are reassigned
assertThat(context.getSplitsAssignmentSequence()).hasSizeGreaterThan(1);
diff --git a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/proxy/KinesisStreamProxyTest.java b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/proxy/KinesisStreamProxyTest.java
index 27bc41c..f53b1ce 100644
--- a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/proxy/KinesisStreamProxyTest.java
+++ b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/proxy/KinesisStreamProxyTest.java
@@ -22,8 +22,6 @@ import org.apache.flink.connector.kinesis.source.split.StartingPosition;
import org.apache.flink.connector.kinesis.source.util.KinesisClientProvider.ListShardItem;
import org.apache.flink.connector.kinesis.source.util.KinesisClientProvider.TestingKinesisClient;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
-
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.NullAndEmptySource;
@@ -42,9 +40,12 @@ import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
import java.time.Instant;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.flink.connector.kinesis.source.util.TestUtil.generateShardId;
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
@@ -62,7 +63,7 @@ class KinesisStreamProxyTest {
final List<Shard> expectedShards = getTestShards(0, 3);
List<ListShardItem> listShardItems =
- ImmutableList.of(
+ Collections.singletonList(
ListShardItem.builder()
.validation(
getListShardRequestValidation(
@@ -88,28 +89,30 @@ class KinesisStreamProxyTest {
final List<Shard> expectedShards = getTestShards(0, 3);
List<ListShardItem> listShardItems =
- ImmutableList.of(
- ListShardItem.builder()
- .validation(
- getListShardRequestValidation(
- streamArn, lastSeenShardId, null))
- .shards(expectedShards.subList(0, 1))
- .nextToken("next-token-1")
- .build(),
- ListShardItem.builder()
- .validation(
- getListShardRequestValidation(
- streamArn, null, "next-token-1"))
- .shards(expectedShards.subList(1, 2))
- .nextToken("next-token-2")
- .build(),
- ListShardItem.builder()
- .validation(
- getListShardRequestValidation(
- streamArn, null, "next-token-2"))
- .shards(expectedShards.subList(2, 4))
- .nextToken(null)
- .build());
+ Stream.of(
+ ListShardItem.builder()
+ .validation(
+ getListShardRequestValidation(
+ streamArn, lastSeenShardId, null))
+ .shards(expectedShards.subList(0, 1))
+ .nextToken("next-token-1")
+ .build(),
+ ListShardItem.builder()
+ .validation(
+ getListShardRequestValidation(
+ streamArn, null, "next-token-1"))
+ .shards(expectedShards.subList(1, 2))
+ .nextToken("next-token-2")
+ .build(),
+ ListShardItem.builder()
+ .validation(
+ getListShardRequestValidation(
+ streamArn, null, "next-token-2"))
+ .shards(expectedShards.subList(2, 4))
+ .nextToken(null)
+ .build())
+ .collect(Collectors.toList());
+
TestingKinesisClient testKinesisClient = new TestingKinesisClient();
testKinesisClient.setListShardsResponses(listShardItems);
diff --git a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/reader/KinesisStreamsRecordEmitterTest.java b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/reader/KinesisStreamsRecordEmitterTest.java
index 680a5a6..f9489f9 100644
--- a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/reader/KinesisStreamsRecordEmitterTest.java
+++ b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/reader/KinesisStreamsRecordEmitterTest.java
@@ -28,8 +28,6 @@ import org.apache.flink.connector.kinesis.source.split.KinesisShardSplitState;
import org.apache.flink.connector.kinesis.source.split.StartingPosition;
import org.apache.flink.util.Collector;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
-
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.services.kinesis.model.Record;
@@ -39,6 +37,8 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.flink.connector.kinesis.source.util.TestUtil.getTestSplitState;
import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat;
@@ -51,20 +51,27 @@ class KinesisStreamsRecordEmitterTest {
void testEmitRecord() throws Exception {
final Instant startTime = Instant.now();
List<Record> inputRecords =
- ImmutableList.of(
- Record.builder()
- .data(SdkBytes.fromByteArray(STRING_SCHEMA.serialize("data-1")))
- .approximateArrivalTimestamp(startTime)
- .build(),
- Record.builder()
- .data(SdkBytes.fromByteArray(STRING_SCHEMA.serialize("data-2")))
- .approximateArrivalTimestamp(startTime.plusSeconds(10))
- .build(),
- Record.builder()
- .data(SdkBytes.fromByteArray(STRING_SCHEMA.serialize("data-3")))
- .approximateArrivalTimestamp(startTime.plusSeconds(20))
- .sequenceNumber("some-sequence-number")
- .build());
+ Stream.of(
+ Record.builder()
+ .data(
+ SdkBytes.fromByteArray(
+ STRING_SCHEMA.serialize("data-1")))
+ .approximateArrivalTimestamp(startTime)
+ .build(),
+ Record.builder()
+ .data(
+ SdkBytes.fromByteArray(
+ STRING_SCHEMA.serialize("data-2")))
+ .approximateArrivalTimestamp(startTime.plusSeconds(10))
+ .build(),
+ Record.builder()
+ .data(
+ SdkBytes.fromByteArray(
+ STRING_SCHEMA.serialize("data-3")))
+ .approximateArrivalTimestamp(startTime.plusSeconds(20))
+ .sequenceNumber("some-sequence-number")
+ .build())
+ .collect(Collectors.toList());
final StartingPosition expectedStartingPosition =
StartingPosition.continueFromSequenceNumber("some-sequence-number");
final CapturingSourceOutput<String> output = new CapturingSourceOutput<>();
@@ -91,22 +98,29 @@ class KinesisStreamsRecordEmitterTest {
void testEmitRecordBasedOnSequenceNumber() throws Exception {
final Instant startTime = Instant.now();
List<Record> inputRecords =
- ImmutableList.of(
- Record.builder()
- .data(SdkBytes.fromByteArray(STRING_SCHEMA.serialize("data-1")))
- .sequenceNumber("emit")
- .approximateArrivalTimestamp(startTime)
- .build(),
- Record.builder()
- .data(SdkBytes.fromByteArray(STRING_SCHEMA.serialize("data-2")))
- .sequenceNumber("emit")
- .approximateArrivalTimestamp(startTime.plusSeconds(10))
- .build(),
- Record.builder()
- .data(SdkBytes.fromByteArray(STRING_SCHEMA.serialize("data-3")))
- .approximateArrivalTimestamp(startTime.plusSeconds(20))
- .sequenceNumber("do-not-emit")
- .build());
+ Stream.of(
+ Record.builder()
+ .data(
+ SdkBytes.fromByteArray(
+ STRING_SCHEMA.serialize("data-1")))
+ .sequenceNumber("emit")
+ .approximateArrivalTimestamp(startTime)
+ .build(),
+ Record.builder()
+ .data(
+ SdkBytes.fromByteArray(
+ STRING_SCHEMA.serialize("data-2")))
+ .sequenceNumber("emit")
+ .approximateArrivalTimestamp(startTime.plusSeconds(10))
+ .build(),
+ Record.builder()
+ .data(
+ SdkBytes.fromByteArray(
+ STRING_SCHEMA.serialize("data-3")))
+ .approximateArrivalTimestamp(startTime.plusSeconds(20))
+ .sequenceNumber("do-not-emit")
+ .build())
+ .collect(Collectors.toList());
final CapturingSourceOutput<String> output = new CapturingSourceOutput<>();
final KinesisShardSplitState splitState = getTestSplitState();
@@ -126,20 +140,27 @@ class KinesisStreamsRecordEmitterTest {
void testEmitRecordWithMetadata() throws Exception {
final Instant startTime = Instant.now();
List<Record> inputRecords =
- ImmutableList.of(
- Record.builder()
- .data(SdkBytes.fromByteArray(STRING_SCHEMA.serialize("data-1")))
- .approximateArrivalTimestamp(startTime)
- .build(),
- Record.builder()
- .data(SdkBytes.fromByteArray(STRING_SCHEMA.serialize("data-2")))
- .approximateArrivalTimestamp(startTime.plusSeconds(10))
- .build(),
- Record.builder()
- .data(SdkBytes.fromByteArray(STRING_SCHEMA.serialize("data-3")))
- .approximateArrivalTimestamp(startTime.plusSeconds(20))
- .sequenceNumber("some-sequence-number")
- .build());
+ Stream.of(
+ Record.builder()
+ .data(
+ SdkBytes.fromByteArray(
+ STRING_SCHEMA.serialize("data-1")))
+ .approximateArrivalTimestamp(startTime)
+ .build(),
+ Record.builder()
+ .data(
+ SdkBytes.fromByteArray(
+ STRING_SCHEMA.serialize("data-2")))
+ .approximateArrivalTimestamp(startTime.plusSeconds(10))
+ .build(),
+ Record.builder()
+ .data(
+ SdkBytes.fromByteArray(
+ STRING_SCHEMA.serialize("data-3")))
+ .approximateArrivalTimestamp(startTime.plusSeconds(20))
+ .sequenceNumber("some-sequence-number")
+ .build())
+ .collect(Collectors.toList());
final CapturingSourceOutput<String> output = new CapturingSourceOutput<>();
final KinesisShardSplitState splitState = getTestSplitState();
diff --git a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/reader/PollingKinesisShardSplitReaderTest.java b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/reader/PollingKinesisShardSplitReaderTest.java
index 5b029cd..1e5c20d 100644
--- a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/reader/PollingKinesisShardSplitReaderTest.java
+++ b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/source/reader/PollingKinesisShardSplitReaderTest.java
@@ -25,14 +25,14 @@ import org.apache.flink.connector.kinesis.source.split.KinesisShardSplit;
import org.apache.flink.connector.kinesis.source.util.KinesisStreamProxyProvider.TestKinesisStreamProxy;
import org.apache.flink.connector.kinesis.source.util.TestUtil;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
-
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.services.kinesis.model.Record;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.flink.connector.kinesis.source.util.KinesisStreamProxyProvider.getTestStreamProxy;
import static org.apache.flink.connector.kinesis.source.util.TestUtil.generateShardId;
@@ -65,7 +65,7 @@ class PollingKinesisShardSplitReaderTest {
String shardId = generateShardId(1);
testStreamProxy.addShards(shardId);
splitReader.handleSplitsChanges(
- new SplitsAddition<>(ImmutableList.of(getTestSplit(shardId))));
+ new SplitsAddition<>(Collections.singletonList(getTestSplit(shardId))));
// When fetching records
RecordsWithSplitIds<Record> retrievedRecords = splitReader.fetch();
@@ -86,16 +86,16 @@ class PollingKinesisShardSplitReaderTest {
String shardId = generateShardId(1);
testStreamProxy.addShards(shardId);
List<Record> expectedRecords =
- ImmutableList.of(
- getTestRecord("data-1"), getTestRecord("data-2"), getTestRecord("data-3"));
+ Stream.of(getTestRecord("data-1"), getTestRecord("data-2"), getTestRecord("data-3"))
+ .collect(Collectors.toList());
testStreamProxy.addRecords(
- TestUtil.STREAM_ARN, shardId, ImmutableList.of(expectedRecords.get(0)));
+ TestUtil.STREAM_ARN, shardId, Collections.singletonList(expectedRecords.get(0)));
testStreamProxy.addRecords(
- TestUtil.STREAM_ARN, shardId, ImmutableList.of(expectedRecords.get(1)));
+ TestUtil.STREAM_ARN, shardId, Collections.singletonList(expectedRecords.get(1)));
testStreamProxy.addRecords(
- TestUtil.STREAM_ARN, shardId, ImmutableList.of(expectedRecords.get(2)));
+ TestUtil.STREAM_ARN, shardId, Collections.singletonList(expectedRecords.get(2)));
splitReader.handleSplitsChanges(
- new SplitsAddition<>(ImmutableList.of(getTestSplit(shardId))));
+ new SplitsAddition<>(Collections.singletonList(getTestSplit(shardId))));
// When fetching records
List<Record> records = new ArrayList<>();
@@ -117,16 +117,16 @@ class PollingKinesisShardSplitReaderTest {
String shardId = generateShardId(1);
testStreamProxy.addShards(shardId);
List<Record> expectedRecords =
- ImmutableList.of(
- getTestRecord("data-1"), getTestRecord("data-2"), getTestRecord("data-3"));
+ Stream.of(getTestRecord("data-1"), getTestRecord("data-2"), getTestRecord("data-3"))
+ .collect(Collectors.toList());
testStreamProxy.addRecords(
- TestUtil.STREAM_ARN, shardId, ImmutableList.of(expectedRecords.get(0)));
+ TestUtil.STREAM_ARN, shardId, Collections.singletonList(expectedRecords.get(0)));
testStreamProxy.addRecords(
- TestUtil.STREAM_ARN, shardId, ImmutableList.of(expectedRecords.get(1)));
+ TestUtil.STREAM_ARN, shardId, Collections.singletonList(expectedRecords.get(1)));
testStreamProxy.addRecords(
- TestUtil.STREAM_ARN, shardId, ImmutableList.of(expectedRecords.get(2)));
+ TestUtil.STREAM_ARN, shardId, Collections.singletonList(expectedRecords.get(2)));
splitReader.handleSplitsChanges(
- new SplitsAddition<>(ImmutableList.of(getTestSplit(shardId))));
+ new SplitsAddition<>(Collections.singletonList(getTestSplit(shardId))));
// When records are fetched
List<Record> fetchedRecords = new ArrayList<>();
@@ -150,7 +150,7 @@ class PollingKinesisShardSplitReaderTest {
testStreamProxy.addShards(shardId);
testStreamProxy.addRecords(TestUtil.STREAM_ARN, shardId, Collections.emptyList());
KinesisShardSplit split = getTestSplit(shardId);
- splitReader.handleSplitsChanges(new SplitsAddition<>(ImmutableList.of(split)));
+ splitReader.handleSplitsChanges(new SplitsAddition<>(Collections.singletonList(split)));
testStreamProxy.setShouldCompleteNextShard(true);
// When fetching records
@@ -172,11 +172,11 @@ class PollingKinesisShardSplitReaderTest {
String shardId = generateShardId(1);
testStreamProxy.addShards(shardId);
List<Record> expectedRecords =
- ImmutableList.of(
- getTestRecord("data-1"), getTestRecord("data-2"), getTestRecord("data-3"));
+ Stream.of(getTestRecord("data-1"), getTestRecord("data-2"), getTestRecord("data-3"))
+ .collect(Collectors.toList());
testStreamProxy.addRecords(TestUtil.STREAM_ARN, shardId, expectedRecords);
KinesisShardSplit split = getTestSplit(shardId);
- splitReader.handleSplitsChanges(new SplitsAddition<>(ImmutableList.of(split)));
+ splitReader.handleSplitsChanges(new SplitsAddition<>(Collections.singletonList(split)));
// When fetching records
List<Record> fetchedRecords = new ArrayList<>();