You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by da...@apache.org on 2022/10/10 13:00:43 UTC
[flink] branch master updated: [FLINK-29395][Connector/Kinesis] Handle empty deaggregated records in FanOutRecordPublisher
This is an automated email from the ASF dual-hosted git repository.
dannycranmer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ef93ae4525e [FLINK-29395][Connector/Kinesis] Handle empty deaggregated records in FanOutRecordPublisher
ef93ae4525e is described below
commit ef93ae4525ea42a87baf77747dd3ffbc007112fd
Author: Hong Liang Teoh <li...@amazon.com>
AuthorDate: Tue Oct 4 08:44:29 2022 +0100
[FLINK-29395][Connector/Kinesis] Handle empty deaggregated records in FanOutRecordPublisher
---
.../publisher/fanout/FanOutRecordPublisher.java | 19 ++++++-
.../fanout/FanOutRecordPublisherTest.java | 59 ++++++++++++++++++++++
.../FakeKinesisFanOutBehavioursFactory.java | 20 ++++++++
.../connectors/kinesis/testutils/TestUtils.java | 26 +++++++---
4 files changed, 116 insertions(+), 8 deletions(-)
diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/FanOutRecordPublisher.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/FanOutRecordPublisher.java
index ef653d71b8b..9df68bf7598 100644
--- a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/FanOutRecordPublisher.java
+++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/FanOutRecordPublisher.java
@@ -44,9 +44,11 @@ import java.util.Date;
import java.util.List;
import java.util.function.Consumer;
+import static com.amazonaws.services.kinesis.model.ShardIteratorType.AT_TIMESTAMP;
import static org.apache.flink.streaming.connectors.kinesis.internals.publisher.RecordPublisher.RecordPublisherRunResult.CANCELLED;
import static org.apache.flink.streaming.connectors.kinesis.internals.publisher.RecordPublisher.RecordPublisherRunResult.COMPLETE;
import static org.apache.flink.streaming.connectors.kinesis.internals.publisher.RecordPublisher.RecordPublisherRunResult.INCOMPLETE;
+import static org.apache.flink.streaming.connectors.kinesis.model.SentinelSequenceNumber.SENTINEL_AT_TIMESTAMP_SEQUENCE_NUM;
import static software.amazon.awssdk.services.kinesis.model.StartingPosition.builder;
/**
@@ -116,8 +118,7 @@ public class FanOutRecordPublisher implements RecordPublisher {
subscribedShard,
event.millisBehindLatest());
SequenceNumber sequenceNumber = recordConsumer.accept(recordBatch);
- nextStartingPosition =
- StartingPosition.continueFromSequenceNumber(sequenceNumber);
+ nextStartingPosition = getNextStartingPosition(sequenceNumber);
};
RecordPublisherRunResult result = runWithBackoff(eventConsumer);
@@ -131,6 +132,20 @@ public class FanOutRecordPublisher implements RecordPublisher {
return result;
}
+ private StartingPosition getNextStartingPosition(final SequenceNumber latestSequenceNumber) {
+ // When consuming from a timestamp sentinel/AT_TIMESTAMP ShardIteratorType.
+ // If the first RecordBatch has no deaggregated records, then the latestSequenceNumber would
+ // be the timestamp sentinel.
+ // This is because we have not yet received any real sequence numbers on this shard.
+ // In this condition we should retry from the previous starting position (AT_TIMESTAMP).
+ if (SENTINEL_AT_TIMESTAMP_SEQUENCE_NUM.get().equals(latestSequenceNumber)) {
+ Preconditions.checkState(nextStartingPosition.getShardIteratorType() == AT_TIMESTAMP);
+ return nextStartingPosition;
+ } else {
+ return StartingPosition.continueFromSequenceNumber(latestSequenceNumber);
+ }
+ }
+
/**
* Runs the record publisher, will sleep for configuration computed jitter period in the case of
* certain exceptions. Unrecoverable exceptions are thrown to terminate the application.
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/FanOutRecordPublisherTest.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/FanOutRecordPublisherTest.java
index 548ef737319..21a695ed412 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/FanOutRecordPublisherTest.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/FanOutRecordPublisherTest.java
@@ -25,12 +25,15 @@ import org.apache.flink.streaming.connectors.kinesis.model.StartingPosition;
import org.apache.flink.streaming.connectors.kinesis.proxy.FullJitterBackoff;
import org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxyV2Interface;
import org.apache.flink.streaming.connectors.kinesis.testutils.FakeKinesisFanOutBehavioursFactory;
+import org.apache.flink.streaming.connectors.kinesis.testutils.FakeKinesisFanOutBehavioursFactory.AbstractSingleShardFanOutKinesisV2;
import org.apache.flink.streaming.connectors.kinesis.testutils.FakeKinesisFanOutBehavioursFactory.SingleShardFanOutKinesisV2;
import org.apache.flink.streaming.connectors.kinesis.testutils.FakeKinesisFanOutBehavioursFactory.SubscriptionErrorKinesisV2;
import org.apache.flink.streaming.connectors.kinesis.testutils.TestUtils.TestConsumer;
import com.amazonaws.http.timers.client.SdkInterruptedException;
+import com.amazonaws.kinesis.agg.RecordAggregator;
import com.amazonaws.services.kinesis.clientlibrary.types.UserRecord;
+import com.amazonaws.services.kinesis.model.HashKeyRange;
import io.netty.handler.timeout.ReadTimeoutException;
import org.junit.Rule;
import org.junit.Test;
@@ -42,12 +45,16 @@ import software.amazon.awssdk.services.kinesis.model.ResourceNotFoundException;
import software.amazon.awssdk.services.kinesis.model.SubscribeToShardEvent;
import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Collections.emptyList;
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.EFO_CONSUMER_NAME;
import static org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.RECORD_PUBLISHER_TYPE;
import static org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.RecordPublisherType.EFO;
@@ -58,9 +65,11 @@ import static org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfi
import static org.apache.flink.streaming.connectors.kinesis.internals.publisher.RecordPublisher.RecordPublisherRunResult.CANCELLED;
import static org.apache.flink.streaming.connectors.kinesis.internals.publisher.RecordPublisher.RecordPublisherRunResult.COMPLETE;
import static org.apache.flink.streaming.connectors.kinesis.internals.publisher.RecordPublisher.RecordPublisherRunResult.INCOMPLETE;
+import static org.apache.flink.streaming.connectors.kinesis.model.SentinelSequenceNumber.SENTINEL_AT_TIMESTAMP_SEQUENCE_NUM;
import static org.apache.flink.streaming.connectors.kinesis.model.SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM;
import static org.apache.flink.streaming.connectors.kinesis.model.SentinelSequenceNumber.SENTINEL_LATEST_SEQUENCE_NUM;
import static org.apache.flink.streaming.connectors.kinesis.testutils.FakeKinesisFanOutBehavioursFactory.emptyShard;
+import static org.apache.flink.streaming.connectors.kinesis.testutils.FakeKinesisFanOutBehavioursFactory.singleShardWithEvents;
import static org.apache.flink.streaming.connectors.kinesis.testutils.FakeKinesisFanOutBehavioursFactory.singletonShard;
import static org.apache.flink.streaming.connectors.kinesis.testutils.TestUtils.createDummyStreamShardHandle;
import static org.assertj.core.api.Assertions.assertThat;
@@ -167,6 +176,56 @@ public class FanOutRecordPublisherTest {
assertThat(kinesis.getStartingPositionForSubscription(0).type()).isEqualTo(AT_TIMESTAMP);
}
+ @Test
+ public void testToSdkV2StartingPositionAtTimestampWithDroppedAggregatedRecord()
+ throws Exception {
+ // Create Aggregate Record with explicit hash keys 0 and 1
+ RecordAggregator recordAggregator = new RecordAggregator();
+ recordAggregator.addUserRecord("pk", "0", randomAlphabetic(32).getBytes(UTF_8));
+ recordAggregator.addUserRecord("pk", "1", randomAlphabetic(32).getBytes(UTF_8));
+ Record record =
+ Record.builder()
+ .approximateArrivalTimestamp(Instant.now())
+ .data(
+ SdkBytes.fromByteArray(
+ recordAggregator.clearAndGet().toRecordBytes()))
+ .sequenceNumber("1")
+ .partitionKey("pk")
+ .build();
+
+ // Create 2 Subscription events for asserting that requested sequence number stays the same
+ List<SubscribeToShardEvent> events = new ArrayList<>();
+ events.add(createSubscribeToShardEvent(record));
+ events.add(createSubscribeToShardEvent(record));
+
+ AbstractSingleShardFanOutKinesisV2 kinesis = singleShardWithEvents(events);
+ Date now = new Date();
+
+ // Create ShardHandle with HashKeyRange excluding single UserRecord with hash key 0
+ HashKeyRange hashKeyRange =
+ new HashKeyRange().withStartingHashKey("1").withEndingHashKey("100");
+
+ RecordPublisher publisher =
+ new FanOutRecordPublisher(
+ StartingPosition.fromTimestamp(now),
+ "arn",
+ createDummyStreamShardHandle("stream-name", "shardId-00000", hashKeyRange),
+ kinesis,
+ createConfiguration(),
+ new FullJitterBackoff());
+ publisher.run(
+ new TestConsumer(SENTINEL_AT_TIMESTAMP_SEQUENCE_NUM.get().getSequenceNumber()));
+
+ // Run a second time to ensure the StartingPosition is still respected, indicating we are
+ // still using the original AT_TIMESTAMP behaviour
+ publisher.run(
+ new TestConsumer(SENTINEL_AT_TIMESTAMP_SEQUENCE_NUM.get().getSequenceNumber()));
+
+ assertThat(kinesis.getStartingPositionForSubscription(1).timestamp())
+ .isEqualTo(now.toInstant());
+ assertThat(kinesis.getStartingPositionForSubscription(1).type()).isEqualTo(AT_TIMESTAMP);
+ }
+
@Test
public void testToSdkV1Records() throws Exception {
Date now = new Date();
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/FakeKinesisFanOutBehavioursFactory.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/FakeKinesisFanOutBehavioursFactory.java
index 17f7d436c9c..a416aefbd15 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/FakeKinesisFanOutBehavioursFactory.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/FakeKinesisFanOutBehavioursFactory.java
@@ -84,6 +84,11 @@ public class FakeKinesisFanOutBehavioursFactory {
return new SingletonEventFanOutKinesisV2(event);
}
+ public static AbstractSingleShardFanOutKinesisV2 singleShardWithEvents(
+ final List<SubscribeToShardEvent> events) {
+ return new EventFanOutKinesisV2(events);
+ }
+
public static SingleShardFanOutKinesisV2 emptyShard() {
return new SingleShardFanOutKinesisV2.Builder().withBatchCount(0).build();
}
@@ -262,6 +267,21 @@ public class FakeKinesisFanOutBehavioursFactory {
}
}
+ private static class EventFanOutKinesisV2 extends AbstractSingleShardFanOutKinesisV2 {
+
+ private final List<SubscribeToShardEvent> events;
+
+ private EventFanOutKinesisV2(List<SubscribeToShardEvent> events) {
+ super(1);
+ this.events = events;
+ }
+
+ @Override
+ List<SubscribeToShardEvent> getEventsToSend() {
+ return events;
+ }
+ }
+
private static class MultipleEventsForSingleRequest extends AbstractSingleShardFanOutKinesisV2 {
private MultipleEventsForSingleRequest() {
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/TestUtils.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/TestUtils.java
index 6d1bb8f62e4..7139d82e43f 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/TestUtils.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/TestUtils.java
@@ -118,18 +118,24 @@ public class TestUtils {
public static StreamShardHandle createDummyStreamShardHandle(
final String streamName, final String shardId) {
+ return createDummyStreamShardHandle(
+ streamName,
+ shardId,
+ new HashKeyRange()
+ .withStartingHashKey("0")
+ .withEndingHashKey(
+ new BigInteger(StringUtils.repeat("FF", 16), 16).toString()));
+ }
+
+ public static StreamShardHandle createDummyStreamShardHandle(
+ final String streamName, final String shardId, final HashKeyRange hashKeyRange) {
final Shard shard =
new Shard()
.withSequenceNumberRange(
new SequenceNumberRange()
.withStartingSequenceNumber("0")
.withEndingSequenceNumber("9999999999999"))
- .withHashKeyRange(
- new HashKeyRange()
- .withStartingHashKey("0")
- .withEndingHashKey(
- new BigInteger(StringUtils.repeat("FF", 16), 16)
- .toString()))
+ .withHashKeyRange(hashKeyRange)
.withShardId(shardId);
return new StreamShardHandle(streamName, shard);
@@ -174,6 +180,14 @@ public class TestUtils {
private final List<RecordBatch> recordBatches = new ArrayList<>();
private String latestSequenceNumber;
+ public TestConsumer() {
+ this(null);
+ }
+
+ public TestConsumer(String latestSequenceNumber) {
+ this.latestSequenceNumber = latestSequenceNumber;
+ }
+
@Override
public SequenceNumber accept(final RecordBatch batch) {
recordBatches.add(batch);