luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Overrode SingleStreamTracker#orphanedInitialPositionInStream() to

Browse source 
match that of the provided `StreamConfig`.
This commit is contained in:
stair 2023-04-23 13:15:09 -04:00
parent eccd6cf2e7
commit ce2604e58e
3 changed files with 32 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamConfig.java
View file

@ -27,5 +27,3 @@ public class StreamConfig {
private final InitialPositionInStreamExtended initialPositionInStreamExtended; private final InitialPositionInStreamExtended initialPositionInStreamExtended;
private String consumerArn; private String consumerArn;
} }

14
amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/SingleStreamTracker.java
View file

@ -45,6 +45,14 @@ public class SingleStreamTracker implements StreamTracker {
private final List<StreamConfig> streamConfigs; private final List<StreamConfig> streamConfigs;
/**
* Cached reference to {@link StreamConfig#initialPositionInStreamExtended()}
* to avoid unnecessary getter invocations.
*/
@EqualsAndHashCode.Exclude
@ToString.Exclude
private final InitialPositionInStreamExtended initialPositionInStream;
public SingleStreamTracker(String streamName) { public SingleStreamTracker(String streamName) {
this(StreamIdentifier.singleStreamInstance(streamName)); this(StreamIdentifier.singleStreamInstance(streamName));
} }
@ -72,6 +80,7 @@ public class SingleStreamTracker implements StreamTracker {
public SingleStreamTracker(@NonNull StreamIdentifier streamIdentifier, @NonNull StreamConfig streamConfig) { public SingleStreamTracker(@NonNull StreamIdentifier streamIdentifier, @NonNull StreamConfig streamConfig) {
this.streamIdentifier = streamIdentifier; this.streamIdentifier = streamIdentifier;
this.streamConfigs = Collections.singletonList(streamConfig); this.streamConfigs = Collections.singletonList(streamConfig);
this.initialPositionInStream = streamConfig.initialPositionInStreamExtended();
} }
@Override @Override
@ -84,6 +93,11 @@ public class SingleStreamTracker implements StreamTracker {
return NO_LEASE_DELETION; return NO_LEASE_DELETION;
} }
@Override
public InitialPositionInStreamExtended orphanedStreamInitialPositionInStream() {
return initialPositionInStream;
}
@Override @Override
public boolean isMultiStream() { public boolean isMultiStream() {
return false; return false;

18
amazon-kinesis-client/src/test/java/software/amazon/kinesis/processor/SingleStreamTrackerTest.java
View file

@ -15,6 +15,8 @@
package software.amazon.kinesis.processor; package software.amazon.kinesis.processor;
import java.util.Date;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNotEquals;
@ -48,6 +50,22 @@ public class SingleStreamTrackerTest {
validate(tracker, expectedPosition); validate(tracker, expectedPosition);
} }
@Test
public void testConsistencyOfInitialPositionInStream() {
for (final InitialPositionInStream position : InitialPositionInStream.values()) {
final InitialPositionInStreamExtended expectedPosition;
if (position == InitialPositionInStream.AT_TIMESTAMP) {
expectedPosition = InitialPositionInStreamExtended.newInitialPositionAtTimestamp(new Date());
} else {
expectedPosition = InitialPositionInStreamExtended.newInitialPosition(position);
}
final StreamTracker tracker = new SingleStreamTracker(STREAM_NAME, expectedPosition);
assertEquals(expectedPosition, tracker.orphanedStreamInitialPositionInStream());
}
}
private static void validate(StreamTracker tracker) { private static void validate(StreamTracker tracker) {
validate(tracker, StreamTracker.DEFAULT_POSITION_IN_STREAM); validate(tracker, StreamTracker.DEFAULT_POSITION_IN_STREAM);
} }