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

Reorganize classes to new packages (#311)

Browse source 
* Initial move of some of the classes

Will fix failure with temp public

* Builds + tests pass

Everything builds, and the tests pass.  Continuing migration

* Broke out the lifecycle-ish classes

* Moved a lot of the checkpoint and lease classes now

* Moved most of the remaining classes.

Still have some more to migrate, but is enough to start progress on
other steps.
This commit is contained in:
Justin Pfifer 2018-03-14 14:53:14 -07:00 committed by Sahil Palvia
parent 9c20d1bb22
commit 59ea602333
212 changed files with 2528 additions and 2302 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
amazon-kinesis-client-multilang/src/main/java/com/amazonaws/services/kinesis/multilang/MessageWriter.java
View file

@ -22,9 +22,9 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import software.amazon.kinesis.lifecycle.ShutdownReason;
import software.amazon.kinesis.lifecycle.InitializationInput;
import software.amazon.kinesis.lifecycle.ProcessRecordsInput;
import com.amazonaws.services.kinesis.multilang.messages.CheckpointMessage;
import com.amazonaws.services.kinesis.multilang.messages.InitializeMessage;
import com.amazonaws.services.kinesis.multilang.messages.Message;

6
amazon-kinesis-client-multilang/src/main/java/com/amazonaws/services/kinesis/multilang/MultiLangDaemon.java
View file

@ -23,9 +23,9 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker;
import software.amazon.kinesis.processor.v2.IRecordProcessorFactory;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import software.amazon.kinesis.coordinator.Worker;
import lombok.extern.slf4j.Slf4j;

2
amazon-kinesis-client-multilang/src/main/java/com/amazonaws/services/kinesis/multilang/MultiLangDaemonConfig.java
View file

@ -19,7 +19,7 @@ import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import com.amazonaws.services.kinesis.multilang.config.KinesisClientLibConfigurator;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

10
amazon-kinesis-client-multilang/src/main/java/com/amazonaws/services/kinesis/multilang/MultiLangProtocol.java
View file

@ -21,11 +21,11 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import software.amazon.kinesis.processor.IRecordProcessorCheckpointer;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import software.amazon.kinesis.lifecycle.ShutdownReason;
import software.amazon.kinesis.lifecycle.InitializationInput;
import software.amazon.kinesis.lifecycle.ProcessRecordsInput;
import com.amazonaws.services.kinesis.multilang.messages.CheckpointMessage;
import com.amazonaws.services.kinesis.multilang.messages.InitializeMessage;
import com.amazonaws.services.kinesis.multilang.messages.Message;

14
amazon-kinesis-client-multilang/src/main/java/com/amazonaws/services/kinesis/multilang/MultiLangRecordProcessor.java
View file

@ -20,13 +20,13 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownInput;
import software.amazon.kinesis.processor.IRecordProcessorCheckpointer;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
import software.amazon.kinesis.processor.v2.IShutdownNotificationAware;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import software.amazon.kinesis.lifecycle.InitializationInput;
import software.amazon.kinesis.lifecycle.ProcessRecordsInput;
import software.amazon.kinesis.lifecycle.ShutdownInput;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;

6
amazon-kinesis-client-multilang/src/main/java/com/amazonaws/services/kinesis/multilang/MultiLangRecordProcessorFactory.java
View file

@ -16,9 +16,9 @@ package com.amazonaws.services.kinesis.multilang;
import java.util.concurrent.ExecutorService;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
import software.amazon.kinesis.processor.v2.IRecordProcessorFactory;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;

2
amazon-kinesis-client-multilang/src/main/java/com/amazonaws/services/kinesis/multilang/config/KinesisClientLibConfigurator.java
View file

@ -29,7 +29,7 @@ import java.util.Set;
import java.util.UUID;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import lombok.extern.slf4j.Slf4j;

2
amazon-kinesis-client-multilang/src/main/java/com/amazonaws/services/kinesis/multilang/messages/InitializeMessage.java
View file

@ -14,7 +14,7 @@
*/
package com.amazonaws.services.kinesis.multilang.messages;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import software.amazon.kinesis.lifecycle.InitializationInput;
import lombok.Getter;
import lombok.Setter;

2
amazon-kinesis-client-multilang/src/main/java/com/amazonaws/services/kinesis/multilang/messages/JsonFriendlyRecord.java
View file

@ -16,7 +16,7 @@ package com.amazonaws.services.kinesis.multilang.messages;
import java.util.Date;
import com.amazonaws.services.kinesis.clientlibrary.types.UserRecord;
import software.amazon.kinesis.retrieval.kpl.UserRecord;
import com.amazonaws.services.kinesis.model.Record;
import com.fasterxml.jackson.annotation.JsonProperty;

2
amazon-kinesis-client-multilang/src/main/java/com/amazonaws/services/kinesis/multilang/messages/ProcessRecordsMessage.java
View file

@ -17,7 +17,7 @@ package com.amazonaws.services.kinesis.multilang.messages;
import java.util.ArrayList;
import java.util.List;
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import software.amazon.kinesis.lifecycle.ProcessRecordsInput;
import com.amazonaws.services.kinesis.model.Record;
import lombok.Getter;
import lombok.Setter;

2
amazon-kinesis-client-multilang/src/main/java/com/amazonaws/services/kinesis/multilang/messages/ShutdownMessage.java
View file

@ -14,7 +14,7 @@
*/
package com.amazonaws.services.kinesis.multilang.messages;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;
import software.amazon.kinesis.lifecycle.ShutdownReason;
/**
* A message to indicate to the client's process that it should shutdown and then terminate.

4
amazon-kinesis-client-multilang/src/test/java/com/amazonaws/services/kinesis/multilang/Matchers.java
View file

@ -21,8 +21,8 @@ import org.hamcrest.Description;
import org.hamcrest.Matcher;
import org.hamcrest.TypeSafeDiagnosingMatcher;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import software.amazon.kinesis.lifecycle.InitializationInput;
public class Matchers {

6
amazon-kinesis-client-multilang/src/test/java/com/amazonaws/services/kinesis/multilang/MessageWriterTest.java
View file

@ -23,14 +23,14 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import software.amazon.kinesis.lifecycle.InitializationInput;
import software.amazon.kinesis.lifecycle.ProcessRecordsInput;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;
import software.amazon.kinesis.lifecycle.ShutdownReason;
import com.amazonaws.services.kinesis.model.Record;
import com.amazonaws.services.kinesis.multilang.messages.Message;
import com.fasterxml.jackson.core.JsonProcessingException;

2
amazon-kinesis-client-multilang/src/test/java/com/amazonaws/services/kinesis/multilang/MultiLangDaemonConfigTest.java
View file

@ -28,7 +28,7 @@ import org.mockito.Mockito;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.services.kinesis.multilang.config.KinesisClientLibConfigurator;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
public class MultiLangDaemonConfigTest {

2
amazon-kinesis-client-multilang/src/test/java/com/amazonaws/services/kinesis/multilang/MultiLangDaemonTest.java
View file

@ -22,7 +22,7 @@ import org.mockito.Mockito;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
public class MultiLangDaemonTest {

10
amazon-kinesis-client-multilang/src/test/java/com/amazonaws/services/kinesis/multilang/MultiLangProtocolTest.java
View file

@ -18,11 +18,11 @@ import com.amazonaws.services.kinesis.clientlibrary.exceptions.InvalidStateExcep
import com.amazonaws.services.kinesis.clientlibrary.exceptions.KinesisClientLibDependencyException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ThrottlingException;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import software.amazon.kinesis.processor.IRecordProcessorCheckpointer;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import software.amazon.kinesis.lifecycle.ShutdownReason;
import software.amazon.kinesis.lifecycle.InitializationInput;
import software.amazon.kinesis.lifecycle.ProcessRecordsInput;
import com.amazonaws.services.kinesis.model.Record;
import com.amazonaws.services.kinesis.multilang.messages.CheckpointMessage;
import com.amazonaws.services.kinesis.multilang.messages.Message;

4
amazon-kinesis-client-multilang/src/test/java/com/amazonaws/services/kinesis/multilang/StreamingRecordProcessorFactoryTest.java
View file

@ -14,11 +14,11 @@
*/
package com.amazonaws.services.kinesis.multilang;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import org.junit.Assert;
import org.junit.Test;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;

14
amazon-kinesis-client-multilang/src/test/java/com/amazonaws/services/kinesis/multilang/StreamingRecordProcessorTest.java
View file

@ -18,13 +18,13 @@ import com.amazonaws.services.kinesis.clientlibrary.exceptions.InvalidStateExcep
import com.amazonaws.services.kinesis.clientlibrary.exceptions.KinesisClientLibDependencyException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ThrottlingException;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IPreparedCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownInput;
import software.amazon.kinesis.processor.IPreparedCheckpointer;
import software.amazon.kinesis.processor.IRecordProcessorCheckpointer;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import software.amazon.kinesis.lifecycle.ShutdownReason;
import software.amazon.kinesis.lifecycle.InitializationInput;
import software.amazon.kinesis.lifecycle.ProcessRecordsInput;
import software.amazon.kinesis.lifecycle.ShutdownInput;
import com.amazonaws.services.kinesis.model.Record;
import com.amazonaws.services.kinesis.multilang.messages.InitializeMessage;
import com.amazonaws.services.kinesis.multilang.messages.Message;

4
amazon-kinesis-client-multilang/src/test/java/com/amazonaws/services/kinesis/multilang/config/KinesisClientLibConfiguratorTest.java
View file

@ -31,8 +31,8 @@ import org.junit.Test;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import software.amazon.kinesis.metrics.MetricsLevel;
import com.google.common.collect.ImmutableSet;
public class KinesisClientLibConfiguratorTest {

6
amazon-kinesis-client-multilang/src/test/java/com/amazonaws/services/kinesis/multilang/messages/MessageTest.java
View file

@ -17,12 +17,12 @@ package com.amazonaws.services.kinesis.multilang.messages;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import software.amazon.kinesis.lifecycle.InitializationInput;
import software.amazon.kinesis.lifecycle.ProcessRecordsInput;
import org.junit.Assert;
import org.junit.Test;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;
import software.amazon.kinesis.lifecycle.ShutdownReason;
import com.amazonaws.services.kinesis.model.Record;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

30
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/interfaces/IRecordProcessorFactory.java
View file

@ -1,30 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.clientlibrary.interfaces;
/**
* The Amazon Kinesis Client Library will use this to instantiate a record processor per shard.
* Clients may choose to create separate instantiations, or re-use instantiations.
*/
public interface IRecordProcessorFactory {
/**
* Returns a record processor to be used for processing data records for a (assigned) shard.
*
* @return Returns a processor object.
*/
IRecordProcessor createProcessor();
}

31
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/interfaces/v2/IRecordProcessorFactory.java
View file

@ -1,31 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.clientlibrary.interfaces.v2;
/**
* The Amazon Kinesis Client Library will use this to instantiate a record processor per shard.
* Clients may choose to create separate instantiations, or re-use instantiations.
*/
public interface IRecordProcessorFactory {
/**
* Returns a record processor to be used for processing data records for a (assigned) shard.
*
* @return Returns a processor object.
*/
IRecordProcessor createProcessor();
}

27
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/checkpoint/Checkpoint.java
View file

@ -1,27 +0,0 @@
package com.amazonaws.services.kinesis.clientlibrary.lib.checkpoint;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import lombok.Data;
/**
* A class encapsulating the 2 pieces of state stored in a checkpoint.
*/
@Data public class Checkpoint {
private final ExtendedSequenceNumber checkpoint;
private final ExtendedSequenceNumber pendingCheckpoint;
/**
* Constructor.
*
* @param checkpoint the checkpoint sequence number - cannot be null or empty.
* @param pendingCheckpoint the pending checkpoint sequence number - can be null.
*/
public Checkpoint(ExtendedSequenceNumber checkpoint, ExtendedSequenceNumber pendingCheckpoint) {
if (checkpoint == null || checkpoint.getSequenceNumber().isEmpty()) {
throw new IllegalArgumentException("Checkpoint cannot be null or empty");
}
this.checkpoint = checkpoint;
this.pendingCheckpoint = pendingCheckpoint;
}
}

7
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/CheckpointValueComparator.java
View file

@ -18,7 +18,8 @@ import java.io.Serializable;
import java.math.BigInteger;
import java.util.Comparator;
import com.amazonaws.services.kinesis.clientlibrary.lib.checkpoint.SentinelCheckpoint;
import software.amazon.kinesis.checkpoint.Checkpoint;
import software.amazon.kinesis.checkpoint.SentinelCheckpoint;
/**
*
@ -87,7 +88,7 @@ class CheckpointValueComparator implements Comparator<String>, Serializable {
* @return a BigInteger value representation of the checkpointValue
*/
private static BigInteger bigIntegerValue(String checkpointValue) {
if (SequenceNumberValidator.isDigits(checkpointValue)) {
if (Checkpoint.SequenceNumberValidator.isDigits(checkpointValue)) {
return new BigInteger(checkpointValue);
} else if (SentinelCheckpoint.LATEST.toString().equals(checkpointValue)) {
return LATEST_BIG_INTEGER_VALUE;
@ -106,7 +107,7 @@ class CheckpointValueComparator implements Comparator<String>, Serializable {
* @return true if and only if the string is all digits or one of the SentinelCheckpoint values
*/
private static boolean isDigitsOrSentinelValue(String string) {
return SequenceNumberValidator.isDigits(string) || isSentinelValue(string);
return Checkpoint.SequenceNumberValidator.isDigits(string) || isSentinelValue(string);
}
/**

8
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/DataFetchingStrategy.java
View file

@ -1,8 +0,0 @@
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
/**
*
*/
public enum DataFetchingStrategy {
DEFAULT, PREFETCH_CACHED;
}

12
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/GetRecordsRetriever.java
View file

@ -1,12 +0,0 @@
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
import com.amazonaws.services.kinesis.model.GetRecordsResult;
import java.util.concurrent.Callable;
/**
* This class uses the GetRecordsRetrievalStrategy class to retrieve the next set of records and update the cache.
*/
public interface GetRecordsRetriever {
GetRecordsResult getNextRecords(int maxRecords);
}

38
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ITask.java
View file

@ -1,38 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.clientlibrary.lib.worker;
import java.util.concurrent.Callable;
/**
* Interface for shard processing tasks.
* A task may execute an application callback (e.g. initialize, process, shutdown).
*/
interface ITask extends Callable<TaskResult> {
/**
* Perform task logic.
* E.g. perform set up (e.g. fetch records) and invoke a callback (e.g. processRecords() API).
*
* @return TaskResult (captures any exceptions encountered during execution of the task)
*/
TaskResult call();
/**
* @return TaskType
*/
TaskType getTaskType();
}

10
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/InitialPositionInStreamExtended.java
View file

@ -20,7 +20,7 @@ import java.util.Date;
* Class that houses the entities needed to specify the position in the stream from where a new application should
* start.
*/
class InitialPositionInStreamExtended {
public class InitialPositionInStreamExtended {
private final InitialPositionInStream position;
private final Date timestamp;
@ -44,7 +44,7 @@ class InitialPositionInStreamExtended {
*
* @return The initial position in stream.
*/
protected InitialPositionInStream getInitialPositionInStream() {
public InitialPositionInStream getInitialPositionInStream() {
return this.position;
}
@ -54,11 +54,11 @@ class InitialPositionInStreamExtended {
*
* @return The timestamp from where we need to start the application.
*/
protected Date getTimestamp() {
public Date getTimestamp() {
return this.timestamp;
}
protected static InitialPositionInStreamExtended newInitialPosition(final InitialPositionInStream position) {
public static InitialPositionInStreamExtended newInitialPosition(final InitialPositionInStream position) {
switch (position) {
case LATEST:
return new InitialPositionInStreamExtended(InitialPositionInStream.LATEST, null);
@ -69,7 +69,7 @@ class InitialPositionInStreamExtended {
}
}
protected static InitialPositionInStreamExtended newInitialPositionAtTimestamp(final Date timestamp) {
public static InitialPositionInStreamExtended newInitialPositionAtTimestamp(final Date timestamp) {
if (timestamp == null) {
throw new IllegalArgumentException("Timestamp must be specified for InitialPosition AT_TIMESTAMP");
}

16
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/NoOpWorkerStateChangeListener.java
View file

@ -1,16 +0,0 @@
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
public class NoOpWorkerStateChangeListener implements WorkerStateChangeListener {
/**
* Empty constructor for NoOp Worker State Change Listener
*/
public NoOpWorkerStateChangeListener() {
}
@Override
public void onWorkerStateChange(WorkerState newState) {
}
}

127
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/SequenceNumberValidator.java
View file

@ -1,127 +0,0 @@
/*
* Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.clientlibrary.lib.worker;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.KinesisClientLibDependencyException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ThrottlingException;
import com.amazonaws.services.kinesis.clientlibrary.proxies.IKinesisProxy;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import com.amazonaws.services.kinesis.model.InvalidArgumentException;
import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
import com.amazonaws.services.kinesis.model.ShardIteratorType;
import lombok.extern.slf4j.Slf4j;
/**
* This class provides some methods for validating sequence numbers. It provides a method
* {@link #validateSequenceNumber(String)} which validates a sequence number by attempting to get an iterator from
* Amazon Kinesis for that sequence number. (e.g. Before checkpointing a client provided sequence number in
* {@link RecordProcessorCheckpointer#checkpoint(String)} to prevent invalid sequence numbers from being checkpointed,
* which could prevent another shard consumer instance from processing the shard later on). This class also provides a
* utility function {@link #isDigits(String)} which is used to check whether a string is all digits
*/
@Slf4j
public class SequenceNumberValidator {
private IKinesisProxy proxy;
private String shardId;
private boolean validateWithGetIterator;
private static final int SERVER_SIDE_ERROR_CODE = 500;
/**
* Constructor.
*
* @param proxy Kinesis proxy to be used for getIterator call
* @param shardId ShardId to check with sequence numbers
* @param validateWithGetIterator Whether to attempt to get an iterator for this shard id and the sequence numbers
* being validated
*/
SequenceNumberValidator(IKinesisProxy proxy, String shardId, boolean validateWithGetIterator) {
this.proxy = proxy;
this.shardId = shardId;
this.validateWithGetIterator = validateWithGetIterator;
}
/**
* Validates the sequence number by attempting to get an iterator from Amazon Kinesis. Repackages exceptions from
* Amazon Kinesis into the appropriate KCL exception to allow clients to determine exception handling strategies
*
* @param sequenceNumber The sequence number to be validated. Must be a numeric string
* @throws IllegalArgumentException Thrown when sequence number validation fails.
* @throws ThrottlingException Thrown when GetShardIterator returns a ProvisionedThroughputExceededException which
* indicates that too many getIterator calls are being made for this shard.
* @throws KinesisClientLibDependencyException Thrown when a service side error is received. This way clients have
* the option of retrying
*/
void validateSequenceNumber(String sequenceNumber)
throws IllegalArgumentException, ThrottlingException, KinesisClientLibDependencyException {
boolean atShardEnd = ExtendedSequenceNumber.SHARD_END.getSequenceNumber().equals(sequenceNumber);
if (!atShardEnd && !isDigits(sequenceNumber)) {
log.info("Sequence number must be numeric, but was {}", sequenceNumber);
throw new IllegalArgumentException("Sequence number must be numeric, but was " + sequenceNumber);
}
try {
if (!atShardEnd &&validateWithGetIterator) {
proxy.getIterator(shardId, ShardIteratorType.AFTER_SEQUENCE_NUMBER.toString(), sequenceNumber);
log.info("Validated sequence number {} with shard id {}", sequenceNumber, shardId);
}
} catch (InvalidArgumentException e) {
log.info("Sequence number {} is invalid for shard {}", sequenceNumber, shardId, e);
throw new IllegalArgumentException("Sequence number " + sequenceNumber + " is invalid for shard "
+ shardId, e);
} catch (ProvisionedThroughputExceededException e) {
// clients should have back off logic in their checkpoint logic
log.info("Exceeded throughput while getting an iterator for shard {}", shardId, e);
throw new ThrottlingException("Exceeded throughput while getting an iterator for shard " + shardId, e);
} catch (AmazonServiceException e) {
log.info("Encountered service exception while getting an iterator for shard {}", shardId, e);
if (e.getStatusCode() >= SERVER_SIDE_ERROR_CODE) {
// clients can choose whether to retry in their checkpoint logic
throw new KinesisClientLibDependencyException("Encountered service exception while getting an iterator"
+ " for shard " + shardId, e);
}
// Just throw any other exceptions, e.g. 400 errors caused by the client
throw e;
}
}
void validateSequenceNumber(ExtendedSequenceNumber checkpoint)
throws IllegalArgumentException, ThrottlingException, KinesisClientLibDependencyException {
validateSequenceNumber(checkpoint.getSequenceNumber());
if (checkpoint.getSubSequenceNumber() < 0) {
throw new IllegalArgumentException("SubSequence number must be non-negative, but was "
+ checkpoint.getSubSequenceNumber());
}
}
/**
* Checks if the string is composed of only digits.
*
* @param string
* @return true for a string of all digits, false otherwise (including false for null and empty string)
*/
static boolean isDigits(String string) {
if (string == null || string.length() == 0) {
return false;
}
for (int i = 0; i < string.length(); i++) {
if (!Character.isDigit(string.charAt(i))) {
return false;
}
}
return true;
}
}

51
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/V1ToV2RecordProcessorAdapter.java
View file

@ -1,51 +0,0 @@
/*
* Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.clientlibrary.lib.worker;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownInput;
/**
* Adapts a V1 {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor IRecordProcessor}
* to V2 {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor IRecordProcessor}.
*/
class V1ToV2RecordProcessorAdapter implements IRecordProcessor {
private com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor recordProcessor;
V1ToV2RecordProcessorAdapter(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor recordProcessor) {
this.recordProcessor = recordProcessor;
}
@Override
public void initialize(InitializationInput initializationInput) {
recordProcessor.initialize(initializationInput.getShardId());
}
@Override
public void processRecords(ProcessRecordsInput processRecordsInput) {
recordProcessor.processRecords(processRecordsInput.getRecords(), processRecordsInput.getCheckpointer());
}
@Override
public void shutdown(ShutdownInput shutdownInput) {
recordProcessor.shutdown(shutdownInput.getCheckpointer(), shutdownInput.getShutdownReason());
}
}

38
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/V1ToV2RecordProcessorFactoryAdapter.java
View file

@ -1,38 +0,0 @@
/*
* Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.clientlibrary.lib.worker;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
/**
* Adapts a V1 {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory
* IRecordProcessorFactory} to V2
* {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory IRecordProcessorFactory}.
*/
class V1ToV2RecordProcessorFactoryAdapter implements IRecordProcessorFactory {
private com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory factory;
V1ToV2RecordProcessorFactoryAdapter(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory factory) {
this.factory = factory;
}
@Override
public IRecordProcessor createProcessor() {
return new V1ToV2RecordProcessorAdapter(factory.createProcessor());
}
}

16
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/WorkerStateChangeListener.java
View file

@ -1,16 +0,0 @@
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
/**
* A listener for callbacks on changes worker state
*/
@FunctionalInterface
public interface WorkerStateChangeListener {
enum WorkerState {
CREATED,
INITIALIZING,
STARTED,
SHUT_DOWN
}
void onWorkerStateChange(WorkerState newState);
}

35
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/proxies/IKinesisProxyExtended.java
View file

@ -1,35 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.clientlibrary.proxies;
import com.amazonaws.services.kinesis.model.Shard;
/**
* Kinesis proxy interface extended with addition method(s). Operates on a
* single stream (set up at initialization).
*
*/
public interface IKinesisProxyExtended extends IKinesisProxy {
/**
* Get the Shard corresponding to shardId associated with this
* IKinesisProxy.
*
* @param shardId
* Fetch the Shard with this given shardId
* @return the Shard with the given shardId
*/
Shard getShard(String shardId);
}

34
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/proxies/IKinesisProxyFactory.java
View file

@ -1,34 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.clientlibrary.proxies;
/**
* Interface for a KinesisProxyFactory.
*
* @deprecated Deprecating since KinesisProxy is just created once, there is no use of a factory. There is no
* replacement for this class. This class will be removed in the next major/minor release.
*
*/
@Deprecated
public interface IKinesisProxyFactory {
/**
* Return an IKinesisProxy object for the specified stream.
* @param streamName Stream from which data is consumed.
* @return IKinesisProxy object.
*/
IKinesisProxy getProxy(String streamName);
}

34
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/exceptions/DependencyException.java
View file

@ -1,34 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.leases.exceptions;
/**
* Indicates that a lease operation has failed because a dependency of the leasing system has failed. This will happen
* if DynamoDB throws an InternalServerException or a generic AmazonClientException (the specific subclasses of
* AmazonClientException are all handled more gracefully).
*/
public class DependencyException extends LeasingException {
private static final long serialVersionUID = 1L;
public DependencyException(Throwable e) {
super(e);
}
public DependencyException(String message, Throwable e) {
super(message, e);
}
}

37
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/exceptions/InvalidStateException.java
View file

@ -1,37 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.leases.exceptions;
/**
* Indicates that a lease operation has failed because DynamoDB is an invalid state. The most common example is failing
* to create the DynamoDB table before doing any lease operations.
*/
public class InvalidStateException extends LeasingException {
private static final long serialVersionUID = 1L;
public InvalidStateException(Throwable e) {
super(e);
}
public InvalidStateException(String message, Throwable e) {
super(message, e);
}
public InvalidStateException(String message) {
super(message);
}
}

36
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/exceptions/LeasingException.java
View file

@ -1,36 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.leases.exceptions;
/**
* Top-level exception type for all exceptions thrown by the leasing code.
*/
public class LeasingException extends Exception {
public LeasingException(Throwable e) {
super(e);
}
public LeasingException(String message, Throwable e) {
super(message, e);
}
public LeasingException(String message) {
super(message);
}
private static final long serialVersionUID = 1L;
}

32
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/exceptions/ProvisionedThroughputException.java
View file

@ -1,32 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.leases.exceptions;
/**
* Indicates that a lease operation has failed due to lack of provisioned throughput for a DynamoDB table.
*/
public class ProvisionedThroughputException extends LeasingException {
private static final long serialVersionUID = 1L;
public ProvisionedThroughputException(Throwable e) {
super(e);
}
public ProvisionedThroughputException(String message, Throwable e) {
super(message, e);
}
}

42
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/interfaces/IKinesisClientLeaseManager.java
View file

@ -1,42 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.leases.interfaces;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import com.amazonaws.services.kinesis.leases.exceptions.DependencyException;
import com.amazonaws.services.kinesis.leases.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.leases.exceptions.ProvisionedThroughputException;
import com.amazonaws.services.kinesis.leases.impl.KinesisClientLease;
/**
* A decoration of ILeaseManager that adds methods to get/update checkpoints.
*/
public interface IKinesisClientLeaseManager extends ILeaseManager<KinesisClientLease> {
/**
* Gets the current checkpoint of the shard. This is useful in the resharding use case
* where we will wait for the parent shard to complete before starting on the records from a child shard.
*
* @param shardId Checkpoint of this shard will be returned
* @return Checkpoint of this shard, or null if the shard record doesn't exist.
*
* @throws ProvisionedThroughputException if DynamoDB update fails due to lack of capacity
* @throws InvalidStateException if lease table does not exist
* @throws DependencyException if DynamoDB update fails in an unexpected way
*/
public abstract ExtendedSequenceNumber getCheckpoint(String shardId)
throws ProvisionedThroughputException, InvalidStateException, DependencyException;
}

29
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/metrics/impl/AccumulateByNameMetricsScope.java
View file

@ -1,29 +0,0 @@
/*
* Copyright 2012-2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.metrics.impl;
/**
* This is a MetricScope with a KeyType of String. It provides the implementation of
* getting the key based off of the String KeyType.
*/
public abstract class AccumulateByNameMetricsScope extends AccumulatingMetricsScope<String> {
@Override
protected String getKey(String name) {
return name;
}
}

29
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/metrics/impl/LogMetricsFactory.java
View file

@ -1,29 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.metrics.impl;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory;
/**
* An IMetricsFactory that creates IMetricsScopes that output themselves via log4j.
*/
public class LogMetricsFactory implements IMetricsFactory {
@Override
public LogMetricsScope createMetrics() {
return new LogMetricsScope();
}
}

29
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/metrics/impl/NullMetricsFactory.java
View file

@ -1,29 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.metrics.impl;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsScope;
public class NullMetricsFactory implements IMetricsFactory {
private static final NullMetricsScope SCOPE = new NullMetricsScope();
@Override
public IMetricsScope createMetrics() {
return SCOPE;
}
}

42
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/metrics/impl/NullMetricsScope.java
View file

@ -1,42 +0,0 @@
/*
* Copyright 2012-2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.metrics.impl;
import com.amazonaws.services.cloudwatch.model.StandardUnit;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsScope;
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel;
public class NullMetricsScope implements IMetricsScope {
@Override
public void addData(String name, double value, StandardUnit unit) {
}
@Override
public void addData(String name, double value, StandardUnit unit, MetricsLevel level) {
}
@Override
public void addDimension(String name, String value) {
}
@Override
public void end() {
}
}

44
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/metrics/impl/ThreadSafeMetricsDelegatingFactory.java
View file

@ -1,44 +0,0 @@
/*
* Copyright 2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.metrics.impl;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsScope;
/**
* Metrics scope factory that delegates metrics scope creation to another factory, but
* returns metrics scope that is thread safe.
*/
public class ThreadSafeMetricsDelegatingFactory implements IMetricsFactory {
/** Metrics factory to delegate to. */
private final IMetricsFactory delegate;
/**
* Creates an instance of the metrics factory.
* @param delegate metrics factory to delegate to
*/
public ThreadSafeMetricsDelegatingFactory(IMetricsFactory delegate) {
this.delegate = delegate;
}
/**
* {@inheritDoc}
*/
@Override
public IMetricsScope createMetrics() {
return new ThreadSafeMetricsDelegatingScope(delegate.createMetrics());
}
}

25
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/metrics/interfaces/IMetricsFactory.java
View file

@ -1,25 +0,0 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 com.amazonaws.services.kinesis.metrics.interfaces;
/**
* Factory for MetricsScope objects.
*/
public interface IMetricsFactory {
/**
* @return a new IMetricsScope object of the type constructed by this factory.
*/
public IMetricsScope createMetrics();
}

151
amazon-kinesis-client/src/main/java/software/amazon/kinesis/checkpoint/Checkpoint.java Normal file
View file

@ -0,0 +1,151 @@
/*
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 software.amazon.kinesis.checkpoint;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.KinesisClientLibDependencyException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ThrottlingException;
import software.amazon.kinesis.coordinator.RecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.model.InvalidArgumentException;
import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
import com.amazonaws.services.kinesis.model.ShardIteratorType;
import lombok.extern.slf4j.Slf4j;
import software.amazon.kinesis.retrieval.IKinesisProxy;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import lombok.Data;
/**
* A class encapsulating the 2 pieces of state stored in a checkpoint.
*/
@Data public class Checkpoint {
private final ExtendedSequenceNumber checkpoint;
private final ExtendedSequenceNumber pendingCheckpoint;
/**
* Constructor.
*
* @param checkpoint the checkpoint sequence number - cannot be null or empty.
* @param pendingCheckpoint the pending checkpoint sequence number - can be null.
*/
public Checkpoint(ExtendedSequenceNumber checkpoint, ExtendedSequenceNumber pendingCheckpoint) {
if (checkpoint == null || checkpoint.getSequenceNumber().isEmpty()) {
throw new IllegalArgumentException("Checkpoint cannot be null or empty");
}
this.checkpoint = checkpoint;
this.pendingCheckpoint = pendingCheckpoint;
}
/**
* This class provides some methods for validating sequence numbers. It provides a method
* {@link #validateSequenceNumber(String)} which validates a sequence number by attempting to get an iterator from
* Amazon Kinesis for that sequence number. (e.g. Before checkpointing a client provided sequence number in
* {@link RecordProcessorCheckpointer#checkpoint(String)} to prevent invalid sequence numbers from being checkpointed,
* which could prevent another shard consumer instance from processing the shard later on). This class also provides a
* utility function {@link #isDigits(String)} which is used to check whether a string is all digits
*/
@Slf4j
public static class SequenceNumberValidator {
private IKinesisProxy proxy;
private String shardId;
private boolean validateWithGetIterator;
private static final int SERVER_SIDE_ERROR_CODE = 500;
/**
* Constructor.
*
* @param proxy Kinesis proxy to be used for getIterator call
* @param shardId ShardId to check with sequence numbers
* @param validateWithGetIterator Whether to attempt to get an iterator for this shard id and the sequence numbers
* being validated
*/
public SequenceNumberValidator(IKinesisProxy proxy, String shardId, boolean validateWithGetIterator) {
this.proxy = proxy;
this.shardId = shardId;
this.validateWithGetIterator = validateWithGetIterator;
}
/**
* Validates the sequence number by attempting to get an iterator from Amazon Kinesis. Repackages exceptions from
* Amazon Kinesis into the appropriate KCL exception to allow clients to determine exception handling strategies
*
* @param sequenceNumber The sequence number to be validated. Must be a numeric string
* @throws IllegalArgumentException Thrown when sequence number validation fails.
* @throws ThrottlingException Thrown when GetShardIterator returns a ProvisionedThroughputExceededException which
* indicates that too many getIterator calls are being made for this shard.
* @throws KinesisClientLibDependencyException Thrown when a service side error is received. This way clients have
* the option of retrying
*/
public void validateSequenceNumber(String sequenceNumber)
throws IllegalArgumentException, ThrottlingException, KinesisClientLibDependencyException {
boolean atShardEnd = ExtendedSequenceNumber.SHARD_END.getSequenceNumber().equals(sequenceNumber);
if (!atShardEnd && !isDigits(sequenceNumber)) {
SequenceNumberValidator.log.info("Sequence number must be numeric, but was {}", sequenceNumber);
throw new IllegalArgumentException("Sequence number must be numeric, but was " + sequenceNumber);
}
try {
if (!atShardEnd &&validateWithGetIterator) {
proxy.getIterator(shardId, ShardIteratorType.AFTER_SEQUENCE_NUMBER.toString(), sequenceNumber);
SequenceNumberValidator.log.info("Validated sequence number {} with shard id {}", sequenceNumber, shardId);
}
} catch (InvalidArgumentException e) {
SequenceNumberValidator.log.info("Sequence number {} is invalid for shard {}", sequenceNumber, shardId, e);
throw new IllegalArgumentException("Sequence number " + sequenceNumber + " is invalid for shard "
+ shardId, e);
} catch (ProvisionedThroughputExceededException e) {
// clients should have back off logic in their checkpoint logic
SequenceNumberValidator.log.info("Exceeded throughput while getting an iterator for shard {}", shardId, e);
throw new ThrottlingException("Exceeded throughput while getting an iterator for shard " + shardId, e);
} catch (AmazonServiceException e) {
SequenceNumberValidator.log.info("Encountered service exception while getting an iterator for shard {}", shardId, e);
if (e.getStatusCode() >= SERVER_SIDE_ERROR_CODE) {
// clients can choose whether to retry in their checkpoint logic
throw new KinesisClientLibDependencyException("Encountered service exception while getting an iterator"
+ " for shard " + shardId, e);
}
// Just throw any other exceptions, e.g. 400 errors caused by the client
throw e;
}
}
void validateSequenceNumber(ExtendedSequenceNumber checkpoint)
throws IllegalArgumentException, ThrottlingException, KinesisClientLibDependencyException {
validateSequenceNumber(checkpoint.getSequenceNumber());
if (checkpoint.getSubSequenceNumber() < 0) {
throw new IllegalArgumentException("SubSequence number must be non-negative, but was "
+ checkpoint.getSubSequenceNumber());
}
}
/**
* Checks if the string is composed of only digits.
*
* @param string
* @return true for a string of all digits, false otherwise (including false for null and empty string)
*/
public static boolean isDigits(String string) {
if (string == null || string.length() == 0) {
return false;
}
for (int i = 0; i < string.length(); i++) {
if (!Character.isDigit(string.charAt(i))) {
return false;
}
}
return true;
}
}
}

6
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/DoesNothingPreparedCheckpointer.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/checkpoint/DoesNothingPreparedCheckpointer.java
View file

@ -12,14 +12,14 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.checkpoint;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.KinesisClientLibDependencyException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ThrottlingException;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IPreparedCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import software.amazon.kinesis.processor.IPreparedCheckpointer;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
/**
* A special IPreparedCheckpointer that does nothing, which can be used when preparing a checkpoint at the current

8
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/PreparedCheckpointer.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/checkpoint/PreparedCheckpointer.java
View file

@ -12,15 +12,15 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.checkpoint;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.KinesisClientLibDependencyException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ThrottlingException;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IPreparedCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import software.amazon.kinesis.processor.IPreparedCheckpointer;
import software.amazon.kinesis.processor.IRecordProcessorCheckpointer;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
/**
* Objects of this class are prepared to checkpoint at a specific sequence number. They use an

4
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/checkpoint/SentinelCheckpoint.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/checkpoint/SentinelCheckpoint.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.checkpoint;
package software.amazon.kinesis.checkpoint;
/**
* Enumeration of the sentinel values of checkpoints.

2
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/GracefulShutdownContext.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/GracefulShutdownContext.java
View file

@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.coordinator;
import lombok.Data;

2
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/GracefulShutdownCoordinator.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/GracefulShutdownCoordinator.java
View file

@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.coordinator;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;

30
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/KinesisClientLibConfiguration.java
View file

@ -12,23 +12,35 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.coordinator;
import java.util.Date;
import java.util.Optional;
import java.util.Set;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStreamExtended;
import org.apache.commons.lang.Validate;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.regions.RegionUtils;
import com.amazonaws.services.kinesis.metrics.impl.MetricsHelper;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsScope;
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel;
import software.amazon.kinesis.leases.NoOpShardPrioritization;
import software.amazon.kinesis.leases.ShardPrioritization;
import software.amazon.kinesis.lifecycle.ProcessRecordsInput;
import software.amazon.kinesis.lifecycle.ProcessTask;
import software.amazon.kinesis.lifecycle.ShardConsumer;
import software.amazon.kinesis.metrics.MetricsHelper;
import software.amazon.kinesis.metrics.IMetricsScope;
import software.amazon.kinesis.metrics.MetricsLevel;
import com.google.common.collect.ImmutableSet;
import lombok.Getter;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
import software.amazon.kinesis.retrieval.DataFetchingStrategy;
import software.amazon.kinesis.retrieval.KinesisProxy;
import software.amazon.kinesis.retrieval.RecordsFetcherFactory;
import software.amazon.kinesis.retrieval.SimpleRecordsFetcherFactory;
/**
* Configuration for the Amazon Kinesis Client Library.
@ -992,13 +1004,13 @@ public class KinesisClientLibConfiguration {
* Controls how long the KCL will sleep if no records are returned from Kinesis
*
* <p>
* This value is only used when no records are returned; if records are returned, the {@link com.amazonaws.services.kinesis.clientlibrary.lib.worker.ProcessTask} will
* This value is only used when no records are returned; if records are returned, the {@link ProcessTask} will
* immediately retrieve the next set of records after the call to
* {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor#processRecords(ProcessRecordsInput)}
* {@link IRecordProcessor#processRecords(ProcessRecordsInput)}
* has returned. Setting this value to high may result in the KCL being unable to catch up. If you are changing this
* value it's recommended that you enable {@link #withCallProcessRecordsEvenForEmptyRecordList(boolean)}, and
* monitor how far behind the records retrieved are by inspecting
* {@link com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput#getMillisBehindLatest()}, and the
* {@link ProcessRecordsInput#getMillisBehindLatest()}, and the
* <a href=
* "http://docs.aws.amazon.com/streams/latest/dev/monitoring-with-cloudwatch.html#kinesis-metrics-stream">CloudWatch
* Metric: GetRecords.MillisBehindLatest</a>
@ -1407,7 +1419,7 @@ public class KinesisClientLibConfiguration {
/**
* @param listShardsBackoffTimeInMillis Max sleep between two listShards call when throttled
* in {@link com.amazonaws.services.kinesis.clientlibrary.proxies.KinesisProxy}.
* in {@link KinesisProxy}.
* @return
*/
public KinesisClientLibConfiguration withListShardsBackoffTimeInMillis(long listShardsBackoffTimeInMillis) {
@ -1418,7 +1430,7 @@ public class KinesisClientLibConfiguration {
/**
* @param maxListShardsRetryAttempts Max number of retries for listShards when throttled
* in {@link com.amazonaws.services.kinesis.clientlibrary.proxies.KinesisProxy}.
* in {@link KinesisProxy}.
* @return
*/
public KinesisClientLibConfiguration withMaxListShardsRetryAttempts(int maxListShardsRetryAttempts) {

30
amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/NoOpWorkerStateChangeListener.java Normal file
View file

@ -0,0 +1,30 @@
/*
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 software.amazon.kinesis.coordinator;
public class NoOpWorkerStateChangeListener implements WorkerStateChangeListener {
/**
* Empty constructor for NoOp Worker State Change Listener
*/
public NoOpWorkerStateChangeListener() {
}
@Override
public void onWorkerStateChange(WorkerState newState) {
}
}

40
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/RecordProcessorCheckpointer.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/RecordProcessorCheckpointer.java
View file

@ -12,21 +12,25 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.coordinator;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.KinesisClientLibDependencyException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.KinesisClientLibException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ThrottlingException;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.ICheckpoint;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IPreparedCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import com.amazonaws.services.kinesis.clientlibrary.types.UserRecord;
import com.amazonaws.services.kinesis.metrics.impl.MetricsHelper;
import com.amazonaws.services.kinesis.metrics.impl.ThreadSafeMetricsDelegatingScope;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory;
import software.amazon.kinesis.checkpoint.Checkpoint;
import software.amazon.kinesis.checkpoint.DoesNothingPreparedCheckpointer;
import software.amazon.kinesis.checkpoint.PreparedCheckpointer;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.processor.ICheckpoint;
import software.amazon.kinesis.processor.IPreparedCheckpointer;
import software.amazon.kinesis.processor.IRecordProcessorCheckpointer;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import software.amazon.kinesis.retrieval.kpl.UserRecord;
import software.amazon.kinesis.metrics.MetricsHelper;
import software.amazon.kinesis.metrics.ThreadSafeMetricsDelegatingScope;
import software.amazon.kinesis.metrics.IMetricsFactory;
import com.amazonaws.services.kinesis.model.Record;
import lombok.extern.slf4j.Slf4j;
@ -37,7 +41,7 @@ import lombok.extern.slf4j.Slf4j;
* RecordProcessor instance. Amazon Kinesis Client Library will create one instance per shard assignment.
*/
@Slf4j
class RecordProcessorCheckpointer implements IRecordProcessorCheckpointer {
public class RecordProcessorCheckpointer implements IRecordProcessorCheckpointer {
private ICheckpoint checkpoint;
private ExtendedSequenceNumber largestPermittedCheckpointValue;
@ -47,7 +51,7 @@ class RecordProcessorCheckpointer implements IRecordProcessorCheckpointer {
private ShardInfo shardInfo;
private SequenceNumberValidator sequenceNumberValidator;
private Checkpoint.SequenceNumberValidator sequenceNumberValidator;
private ExtendedSequenceNumber sequenceNumberAtShardEnd;
@ -59,9 +63,9 @@ class RecordProcessorCheckpointer implements IRecordProcessorCheckpointer {
* @param checkpoint Used to checkpoint progress of a RecordProcessor
* @param validator Used for validating sequence numbers
*/
RecordProcessorCheckpointer(ShardInfo shardInfo,
public RecordProcessorCheckpointer(ShardInfo shardInfo,
ICheckpoint checkpoint,
SequenceNumberValidator validator,
Checkpoint.SequenceNumberValidator validator,
IMetricsFactory metricsFactory) {
this.shardInfo = shardInfo;
this.checkpoint = checkpoint;
@ -227,11 +231,11 @@ class RecordProcessorCheckpointer implements IRecordProcessorCheckpointer {
/**
* @return the lastCheckpointValue
*/
ExtendedSequenceNumber getLastCheckpointValue() {
public ExtendedSequenceNumber getLastCheckpointValue() {
return lastCheckpointValue;
}
synchronized void setInitialCheckpointValue(ExtendedSequenceNumber initialCheckpoint) {
public synchronized void setInitialCheckpointValue(ExtendedSequenceNumber initialCheckpoint) {
lastCheckpointValue = initialCheckpoint;
}
@ -240,14 +244,14 @@ class RecordProcessorCheckpointer implements IRecordProcessorCheckpointer {
*
* @return the largest permitted checkpoint
*/
synchronized ExtendedSequenceNumber getLargestPermittedCheckpointValue() {
public synchronized ExtendedSequenceNumber getLargestPermittedCheckpointValue() {
return largestPermittedCheckpointValue;
}
/**
* @param largestPermittedCheckpointValue the largest permitted checkpoint
*/
synchronized void setLargestPermittedCheckpointValue(ExtendedSequenceNumber largestPermittedCheckpointValue) {
public synchronized void setLargestPermittedCheckpointValue(ExtendedSequenceNumber largestPermittedCheckpointValue) {
this.largestPermittedCheckpointValue = largestPermittedCheckpointValue;
}
@ -258,7 +262,7 @@ class RecordProcessorCheckpointer implements IRecordProcessorCheckpointer {
*
* @param extendedSequenceNumber
*/
synchronized void setSequenceNumberAtShardEnd(ExtendedSequenceNumber extendedSequenceNumber) {
public synchronized void setSequenceNumberAtShardEnd(ExtendedSequenceNumber extendedSequenceNumber) {
this.sequenceNumberAtShardEnd = extendedSequenceNumber;
}

23
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/StreamConfig.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/StreamConfig.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,14 +12,15 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.coordinator;
import com.amazonaws.services.kinesis.clientlibrary.proxies.IKinesisProxy;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStreamExtended;
import software.amazon.kinesis.retrieval.IKinesisProxy;
/**
* Used to capture stream configuration and pass it along.
*/
class StreamConfig {
public class StreamConfig {
private final IKinesisProxy streamProxy;
private final int maxRecords;
@ -37,7 +38,7 @@ class StreamConfig {
* @param validateSequenceNumberBeforeCheckpointing Whether to call Amazon Kinesis to validate sequence numbers
* @param initialPositionInStream Initial position in stream
*/
StreamConfig(IKinesisProxy proxy,
public StreamConfig(IKinesisProxy proxy,
int maxRecords,
long idleTimeInMilliseconds,
boolean callProcessRecordsEvenForEmptyRecordList,
@ -54,42 +55,42 @@ class StreamConfig {
/**
* @return the streamProxy
*/
IKinesisProxy getStreamProxy() {
public IKinesisProxy getStreamProxy() {
return streamProxy;
}
/**
* @return the maxRecords
*/
int getMaxRecords() {
public int getMaxRecords() {
return maxRecords;
}
/**
* @return the idleTimeInMilliseconds
*/
long getIdleTimeInMilliseconds() {
public long getIdleTimeInMilliseconds() {
return idleTimeInMilliseconds;
}
/**
* @return the callProcessRecordsEvenForEmptyRecordList
*/
boolean shouldCallProcessRecordsEvenForEmptyRecordList() {
public boolean shouldCallProcessRecordsEvenForEmptyRecordList() {
return callProcessRecordsEvenForEmptyRecordList;
}
/**
* @return the initialPositionInStream
*/
InitialPositionInStreamExtended getInitialPositionInStream() {
public InitialPositionInStreamExtended getInitialPositionInStream() {
return initialPositionInStream;
}
/**
* @return validateSequenceNumberBeforeCheckpointing
*/
boolean shouldValidateSequenceNumberBeforeCheckpointing() {
public boolean shouldValidateSequenceNumberBeforeCheckpointing() {
return validateSequenceNumberBeforeCheckpointing;
}
}

70
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/Worker.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/Worker.java
View file

@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.coordinator;
import java.util.Collection;
import java.util.HashSet;
@ -40,20 +40,34 @@ import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;
import com.amazonaws.services.kinesis.AmazonKinesis;
import com.amazonaws.services.kinesis.AmazonKinesisClient;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.ICheckpoint;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware;
import com.amazonaws.services.kinesis.clientlibrary.proxies.IKinesisProxy;
import com.amazonaws.services.kinesis.clientlibrary.proxies.KinesisProxy;
import com.amazonaws.services.kinesis.leases.exceptions.LeasingException;
import com.amazonaws.services.kinesis.leases.impl.KinesisClientLease;
import com.amazonaws.services.kinesis.leases.impl.KinesisClientLeaseManager;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseManager;
import com.amazonaws.services.kinesis.metrics.impl.CWMetricsFactory;
import com.amazonaws.services.kinesis.metrics.impl.NullMetricsFactory;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory;
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStreamExtended;
import software.amazon.kinesis.leases.KinesisClientLibLeaseCoordinator;
import software.amazon.kinesis.leases.ParentsFirstShardPrioritization;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.leases.ShardPrioritization;
import software.amazon.kinesis.leases.ShardSyncTask;
import software.amazon.kinesis.leases.ShardSyncTaskManager;
import software.amazon.kinesis.lifecycle.ShardConsumer;
import software.amazon.kinesis.lifecycle.ShardConsumerShutdownNotification;
import software.amazon.kinesis.lifecycle.ShutdownNotification;
import software.amazon.kinesis.lifecycle.ShutdownReason;
import software.amazon.kinesis.lifecycle.TaskResult;
import software.amazon.kinesis.metrics.MetricsCollectingTaskDecorator;
import software.amazon.kinesis.processor.ICheckpoint;
import software.amazon.kinesis.processor.V1ToV2RecordProcessorFactoryAdapter;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
import software.amazon.kinesis.processor.v2.IRecordProcessorFactory;
import software.amazon.kinesis.processor.v2.IShutdownNotificationAware;
import software.amazon.kinesis.retrieval.IKinesisProxy;
import software.amazon.kinesis.retrieval.KinesisProxy;
import software.amazon.kinesis.leases.exceptions.LeasingException;
import software.amazon.kinesis.leases.KinesisClientLease;
import software.amazon.kinesis.leases.KinesisClientLeaseManager;
import software.amazon.kinesis.leases.ILeaseManager;
import software.amazon.kinesis.metrics.CWMetricsFactory;
import software.amazon.kinesis.metrics.NullMetricsFactory;
import software.amazon.kinesis.metrics.IMetricsFactory;
import software.amazon.kinesis.metrics.MetricsLevel;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
@ -132,7 +146,7 @@ public class Worker implements Runnable {
*/
@Deprecated
public Worker(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory recordProcessorFactory,
software.amazon.kinesis.processor.IRecordProcessorFactory recordProcessorFactory,
KinesisClientLibConfiguration config) {
this(recordProcessorFactory, config, getExecutorService());
}
@ -152,7 +166,7 @@ public class Worker implements Runnable {
*/
@Deprecated
public Worker(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory recordProcessorFactory,
software.amazon.kinesis.processor.IRecordProcessorFactory recordProcessorFactory,
KinesisClientLibConfiguration config, ExecutorService execService) {
this(recordProcessorFactory, config,
new AmazonKinesisClient(config.getKinesisCredentialsProvider(), config.getKinesisClientConfiguration()),
@ -176,7 +190,7 @@ public class Worker implements Runnable {
*/
@Deprecated
public Worker(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory recordProcessorFactory,
software.amazon.kinesis.processor.IRecordProcessorFactory recordProcessorFactory,
KinesisClientLibConfiguration config, IMetricsFactory metricsFactory) {
this(recordProcessorFactory, config, metricsFactory, getExecutorService());
}
@ -196,7 +210,7 @@ public class Worker implements Runnable {
*/
@Deprecated
public Worker(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory recordProcessorFactory,
software.amazon.kinesis.processor.IRecordProcessorFactory recordProcessorFactory,
KinesisClientLibConfiguration config, IMetricsFactory metricsFactory, ExecutorService execService) {
this(recordProcessorFactory, config,
new AmazonKinesisClient(config.getKinesisCredentialsProvider(), config.getKinesisClientConfiguration()),
@ -222,7 +236,7 @@ public class Worker implements Runnable {
*/
@Deprecated
public Worker(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory recordProcessorFactory,
software.amazon.kinesis.processor.IRecordProcessorFactory recordProcessorFactory,
KinesisClientLibConfiguration config, AmazonKinesis kinesisClient, AmazonDynamoDB dynamoDBClient,
AmazonCloudWatch cloudWatchClient) {
this(recordProcessorFactory, config, kinesisClient, dynamoDBClient, cloudWatchClient, getExecutorService());
@ -247,7 +261,7 @@ public class Worker implements Runnable {
*/
@Deprecated
public Worker(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory recordProcessorFactory,
software.amazon.kinesis.processor.IRecordProcessorFactory recordProcessorFactory,
KinesisClientLibConfiguration config, AmazonKinesis kinesisClient, AmazonDynamoDB dynamoDBClient,
AmazonCloudWatch cloudWatchClient, ExecutorService execService) {
this(recordProcessorFactory, config, kinesisClient, dynamoDBClient, getMetricsFactory(cloudWatchClient, config),
@ -275,7 +289,7 @@ public class Worker implements Runnable {
*/
@Deprecated
public Worker(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory recordProcessorFactory,
software.amazon.kinesis.processor.IRecordProcessorFactory recordProcessorFactory,
KinesisClientLibConfiguration config, AmazonKinesisClient kinesisClient,
AmazonDynamoDBClient dynamoDBClient, AmazonCloudWatchClient cloudWatchClient) {
this(recordProcessorFactory, config, (AmazonKinesis) kinesisClient, (AmazonDynamoDB) dynamoDBClient,
@ -304,7 +318,7 @@ public class Worker implements Runnable {
*/
@Deprecated
public Worker(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory recordProcessorFactory,
software.amazon.kinesis.processor.IRecordProcessorFactory recordProcessorFactory,
KinesisClientLibConfiguration config, AmazonKinesisClient kinesisClient,
AmazonDynamoDBClient dynamoDBClient, AmazonCloudWatchClient cloudWatchClient, ExecutorService execService) {
this(recordProcessorFactory, config, (AmazonKinesis) kinesisClient, (AmazonDynamoDB) dynamoDBClient,
@ -333,7 +347,7 @@ public class Worker implements Runnable {
*/
@Deprecated
public Worker(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory recordProcessorFactory,
software.amazon.kinesis.processor.IRecordProcessorFactory recordProcessorFactory,
KinesisClientLibConfiguration config, AmazonKinesisClient kinesisClient,
AmazonDynamoDBClient dynamoDBClient, IMetricsFactory metricsFactory, ExecutorService execService) {
this(recordProcessorFactory, config, (AmazonKinesis) kinesisClient, (AmazonDynamoDB) dynamoDBClient,
@ -359,7 +373,7 @@ public class Worker implements Runnable {
*/
@Deprecated
public Worker(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory recordProcessorFactory,
software.amazon.kinesis.processor.IRecordProcessorFactory recordProcessorFactory,
KinesisClientLibConfiguration config, AmazonKinesis kinesisClient, AmazonDynamoDB dynamoDBClient,
IMetricsFactory metricsFactory, ExecutorService execService) {
this(config.getApplicationName(), new V1ToV2RecordProcessorFactoryAdapter(recordProcessorFactory),
@ -1151,7 +1165,7 @@ public class Worker implements Runnable {
private WorkerStateChangeListener workerStateChangeListener;
/**
* Provide a V1 {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor
* Provide a V1 {@link software.amazon.kinesis.processor.IRecordProcessor
* IRecordProcessor}.
*
* @param recordProcessorFactory
@ -1159,13 +1173,13 @@ public class Worker implements Runnable {
* @return A reference to this updated object so that method calls can be chained together.
*/
public Builder recordProcessorFactory(
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory recordProcessorFactory) {
software.amazon.kinesis.processor.IRecordProcessorFactory recordProcessorFactory) {
this.recordProcessorFactory = new V1ToV2RecordProcessorFactoryAdapter(recordProcessorFactory);
return this;
}
/**
* Provide a V2 {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor
* Provide a V2 {@link IRecordProcessor
* IRecordProcessor}.
*
* @param recordProcessorFactory

30
amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/WorkerStateChangeListener.java Normal file
View file

@ -0,0 +1,30 @@
/*
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 software.amazon.kinesis.coordinator;
/**
* A listener for callbacks on changes worker state
*/
@FunctionalInterface
public interface WorkerStateChangeListener {
enum WorkerState {
CREATED,
INITIALIZING,
STARTED,
SHUT_DOWN
}
void onWorkerStateChange(WorkerState newState);
}

4
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/util/DynamoUtils.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/DynamoUtils.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.util;
package software.amazon.kinesis.leases;
import java.util.ArrayList;
import java.util.Collection;

41
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/IKinesisClientLeaseManager.java Normal file
View file

@ -0,0 +1,41 @@
/*
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 software.amazon.kinesis.leases;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import software.amazon.kinesis.leases.exceptions.DependencyException;
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
import software.amazon.kinesis.leases.exceptions.ProvisionedThroughputException;
/**
* A decoration of ILeaseManager that adds methods to get/update checkpoints.
*/
public interface IKinesisClientLeaseManager extends ILeaseManager<KinesisClientLease> {
/**
* Gets the current checkpoint of the shard. This is useful in the resharding use case
* where we will wait for the parent shard to complete before starting on the records from a child shard.
*
* @param shardId Checkpoint of this shard will be returned
* @return Checkpoint of this shard, or null if the shard record doesn't exist.
*
* @throws ProvisionedThroughputException if DynamoDB update fails due to lack of capacity
* @throws InvalidStateException if lease table does not exist
* @throws DependencyException if DynamoDB update fails in an unexpected way
*/
public abstract ExtendedSequenceNumber getCheckpoint(String shardId)
throws ProvisionedThroughputException, InvalidStateException, DependencyException;
}

12
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/interfaces/ILeaseManager.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ILeaseManager.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,14 +12,14 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.interfaces;
package software.amazon.kinesis.leases;
import java.util.List;
import com.amazonaws.services.kinesis.leases.exceptions.DependencyException;
import com.amazonaws.services.kinesis.leases.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.leases.exceptions.ProvisionedThroughputException;
import com.amazonaws.services.kinesis.leases.impl.Lease;
import software.amazon.kinesis.leases.exceptions.DependencyException;
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
import software.amazon.kinesis.leases.exceptions.ProvisionedThroughputException;
import software.amazon.kinesis.leases.Lease;
/**
* Supports basic CRUD operations for Leases.

12
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/interfaces/ILeaseRenewer.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ILeaseRenewer.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,16 +12,16 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.interfaces;
package software.amazon.kinesis.leases;
import java.util.Collection;
import java.util.Map;
import java.util.UUID;
import com.amazonaws.services.kinesis.leases.exceptions.DependencyException;
import com.amazonaws.services.kinesis.leases.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.leases.exceptions.ProvisionedThroughputException;
import com.amazonaws.services.kinesis.leases.impl.Lease;
import software.amazon.kinesis.leases.exceptions.DependencyException;
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
import software.amazon.kinesis.leases.exceptions.ProvisionedThroughputException;
import software.amazon.kinesis.leases.Lease;
/**
* ILeaseRenewer objects are used by LeaseCoordinator to renew leases held by the LeaseCoordinator. Each

6
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/interfaces/ILeaseSerializer.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ILeaseSerializer.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.interfaces;
package software.amazon.kinesis.leases;
import java.util.Collection;
import java.util.Map;
@ -22,7 +22,7 @@ import com.amazonaws.services.dynamodbv2.model.AttributeValue;
import com.amazonaws.services.dynamodbv2.model.AttributeValueUpdate;
import com.amazonaws.services.dynamodbv2.model.ExpectedAttributeValue;
import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
import com.amazonaws.services.kinesis.leases.impl.Lease;
import software.amazon.kinesis.leases.Lease;
/**
* Utility class that manages the mapping of Lease objects/operations to records in DynamoDB.

10
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/interfaces/ILeaseTaker.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ILeaseTaker.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,13 +12,13 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.interfaces;
package software.amazon.kinesis.leases;
import java.util.Map;
import com.amazonaws.services.kinesis.leases.exceptions.DependencyException;
import com.amazonaws.services.kinesis.leases.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.leases.impl.Lease;
import software.amazon.kinesis.leases.exceptions.DependencyException;
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
import software.amazon.kinesis.leases.Lease;
/**
* ILeaseTaker is used by LeaseCoordinator to take new leases, or leases that other workers fail to renew. Each

4
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/impl/KinesisClientLease.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/KinesisClientLease.java
View file

@ -12,14 +12,14 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.impl;
package software.amazon.kinesis.leases;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
/**
* A Lease subclass containing KinesisClientLibrary related fields for checkpoints.

13
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/impl/KinesisClientLeaseManager.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/KinesisClientLeaseManager.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,14 +12,13 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.impl;
package software.amazon.kinesis.leases;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import com.amazonaws.services.kinesis.leases.exceptions.DependencyException;
import com.amazonaws.services.kinesis.leases.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.leases.exceptions.ProvisionedThroughputException;
import com.amazonaws.services.kinesis.leases.interfaces.IKinesisClientLeaseManager;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import software.amazon.kinesis.leases.exceptions.DependencyException;
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
import software.amazon.kinesis.leases.exceptions.ProvisionedThroughputException;
/**
* An implementation of LeaseManager for the KinesisClientLibrary - takeLease updates the ownerSwitchesSinceCheckpoint field.

6
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/impl/KinesisClientLeaseSerializer.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/KinesisClientLeaseSerializer.java
View file

@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.impl;
package software.amazon.kinesis.leases;
import java.util.Collection;
import java.util.Map;
@ -23,9 +23,7 @@ import com.amazonaws.services.dynamodbv2.model.AttributeValue;
import com.amazonaws.services.dynamodbv2.model.AttributeValueUpdate;
import com.amazonaws.services.dynamodbv2.model.ExpectedAttributeValue;
import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseSerializer;
import com.amazonaws.services.kinesis.leases.util.DynamoUtils;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import com.google.common.base.Strings;
/**

33
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibLeaseCoordinator.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/KinesisClientLibLeaseCoordinator.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.leases;
import java.util.ArrayList;
import java.util.Collection;
@ -27,16 +27,13 @@ import com.amazonaws.services.kinesis.clientlibrary.exceptions.KinesisClientLibE
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ThrottlingException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.internal.KinesisClientLibIOException;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.ICheckpoint;
import com.amazonaws.services.kinesis.clientlibrary.lib.checkpoint.Checkpoint;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import com.amazonaws.services.kinesis.leases.exceptions.DependencyException;
import com.amazonaws.services.kinesis.leases.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.leases.exceptions.ProvisionedThroughputException;
import com.amazonaws.services.kinesis.leases.impl.KinesisClientLease;
import com.amazonaws.services.kinesis.leases.impl.LeaseCoordinator;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseManager;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory;
import software.amazon.kinesis.processor.ICheckpoint;
import software.amazon.kinesis.checkpoint.Checkpoint;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import software.amazon.kinesis.leases.exceptions.DependencyException;
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
import software.amazon.kinesis.leases.exceptions.ProvisionedThroughputException;
import software.amazon.kinesis.metrics.IMetricsFactory;
import lombok.extern.slf4j.Slf4j;
@ -44,7 +41,7 @@ import lombok.extern.slf4j.Slf4j;
* This class is used to coordinate/manage leases owned by this worker process and to get/set checkpoints.
*/
@Slf4j
class KinesisClientLibLeaseCoordinator extends LeaseCoordinator<KinesisClientLease> implements ICheckpoint {
public class KinesisClientLibLeaseCoordinator extends LeaseCoordinator<KinesisClientLease> implements ICheckpoint {
private static final long DEFAULT_INITIAL_LEASE_TABLE_READ_CAPACITY = 10L;
private static final long DEFAULT_INITIAL_LEASE_TABLE_WRITE_CAPACITY = 10L;
@ -144,7 +141,7 @@ class KinesisClientLibLeaseCoordinator extends LeaseCoordinator<KinesisClientLea
* @throws ProvisionedThroughputException if DynamoDB update fails due to lack of capacity
* @throws DependencyException if DynamoDB update fails in an unexpected way
*/
boolean setCheckpoint(String shardId, ExtendedSequenceNumber checkpoint, UUID concurrencyToken)
public boolean setCheckpoint(String shardId, ExtendedSequenceNumber checkpoint, UUID concurrencyToken)
throws DependencyException, InvalidStateException, ProvisionedThroughputException {
KinesisClientLease lease = getCurrentlyHeldLease(shardId);
if (lease == null) {
@ -295,7 +292,7 @@ class KinesisClientLibLeaseCoordinator extends LeaseCoordinator<KinesisClientLea
* @throws DependencyException
* @throws ProvisionedThroughputException
*/
void initialize() throws ProvisionedThroughputException, DependencyException, IllegalStateException {
public void initialize() throws ProvisionedThroughputException, DependencyException, IllegalStateException {
final boolean newTableCreated =
leaseManager.createLeaseTableIfNotExists(initialLeaseTableReadCapacity, initialLeaseTableWriteCapacity);
if (newTableCreated) {
@ -317,7 +314,7 @@ class KinesisClientLibLeaseCoordinator extends LeaseCoordinator<KinesisClientLea
* @throws DependencyException
* @throws InvalidStateException
*/
void runLeaseTaker() throws DependencyException, InvalidStateException {
public void runLeaseTaker() throws DependencyException, InvalidStateException {
super.runTaker();
}
@ -327,7 +324,7 @@ class KinesisClientLibLeaseCoordinator extends LeaseCoordinator<KinesisClientLea
* @throws DependencyException
* @throws InvalidStateException
*/
void runLeaseRenewer() throws DependencyException, InvalidStateException {
public void runLeaseRenewer() throws DependencyException, InvalidStateException {
super.runRenewer();
}
@ -337,7 +334,7 @@ class KinesisClientLibLeaseCoordinator extends LeaseCoordinator<KinesisClientLea
*
* @return LeaseManager
*/
ILeaseManager<KinesisClientLease> getLeaseManager() {
public ILeaseManager<KinesisClientLease> getLeaseManager() {
return leaseManager;
}

4
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/impl/Lease.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.impl;
package software.amazon.kinesis.leases;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

27
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/impl/LeaseCoordinator.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseCoordinator.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.impl;
package software.amazon.kinesis.leases;
import java.util.Collection;
import java.util.Map;
@ -26,19 +26,16 @@ import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import com.amazonaws.services.kinesis.leases.exceptions.DependencyException;
import com.amazonaws.services.kinesis.leases.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.leases.exceptions.LeasingException;
import com.amazonaws.services.kinesis.leases.exceptions.ProvisionedThroughputException;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseManager;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseRenewer;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseTaker;
import com.amazonaws.services.kinesis.metrics.impl.LogMetricsFactory;
import com.amazonaws.services.kinesis.metrics.impl.MetricsHelper;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsScope;
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import software.amazon.kinesis.leases.exceptions.DependencyException;
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
import software.amazon.kinesis.leases.exceptions.LeasingException;
import software.amazon.kinesis.leases.exceptions.ProvisionedThroughputException;
import software.amazon.kinesis.metrics.LogMetricsFactory;
import software.amazon.kinesis.metrics.MetricsHelper;
import software.amazon.kinesis.metrics.IMetricsFactory;
import software.amazon.kinesis.metrics.IMetricsScope;
import software.amazon.kinesis.metrics.MetricsLevel;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;

10
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/impl/LeaseManager.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManager.java
View file

@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.impl;
package software.amazon.kinesis.leases;
import java.util.ArrayList;
import java.util.List;
@ -40,11 +40,9 @@ import com.amazonaws.services.dynamodbv2.model.ScanRequest;
import com.amazonaws.services.dynamodbv2.model.ScanResult;
import com.amazonaws.services.dynamodbv2.model.TableStatus;
import com.amazonaws.services.dynamodbv2.model.UpdateItemRequest;
import com.amazonaws.services.kinesis.leases.exceptions.DependencyException;
import com.amazonaws.services.kinesis.leases.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.leases.exceptions.ProvisionedThroughputException;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseManager;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseSerializer;
import software.amazon.kinesis.leases.exceptions.DependencyException;
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
import software.amazon.kinesis.leases.exceptions.ProvisionedThroughputException;
import lombok.extern.slf4j.Slf4j;

20
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/impl/LeaseRenewer.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseRenewer.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.impl;
package software.amazon.kinesis.leases;
import java.util.ArrayList;
import java.util.Collection;
@ -30,15 +30,13 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import com.amazonaws.services.cloudwatch.model.StandardUnit;
import com.amazonaws.services.kinesis.leases.exceptions.DependencyException;
import com.amazonaws.services.kinesis.leases.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.leases.exceptions.ProvisionedThroughputException;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseManager;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseRenewer;
import com.amazonaws.services.kinesis.metrics.impl.MetricsHelper;
import com.amazonaws.services.kinesis.metrics.impl.ThreadSafeMetricsDelegatingScope;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsScope;
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel;
import software.amazon.kinesis.leases.exceptions.DependencyException;
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
import software.amazon.kinesis.leases.exceptions.ProvisionedThroughputException;
import software.amazon.kinesis.metrics.MetricsHelper;
import software.amazon.kinesis.metrics.ThreadSafeMetricsDelegatingScope;
import software.amazon.kinesis.metrics.IMetricsScope;
import software.amazon.kinesis.metrics.MetricsLevel;
import lombok.extern.slf4j.Slf4j;

6
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/impl/LeaseSerializer.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseSerializer.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.impl;
package software.amazon.kinesis.leases;
import java.util.ArrayList;
import java.util.Collection;
@ -28,8 +28,6 @@ import com.amazonaws.services.dynamodbv2.model.ExpectedAttributeValue;
import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
import com.amazonaws.services.dynamodbv2.model.KeyType;
import com.amazonaws.services.dynamodbv2.model.ScalarAttributeType;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseSerializer;
import com.amazonaws.services.kinesis.leases.util.DynamoUtils;
/**
* An implementation of ILeaseSerializer for basic Lease objects. Can also instantiate subclasses of Lease so that

18
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/leases/impl/LeaseTaker.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseTaker.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.leases.impl;
package software.amazon.kinesis.leases;
import java.util.ArrayList;
import java.util.Collection;
@ -27,14 +27,12 @@ import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import com.amazonaws.services.cloudwatch.model.StandardUnit;
import com.amazonaws.services.kinesis.leases.exceptions.DependencyException;
import com.amazonaws.services.kinesis.leases.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.leases.exceptions.ProvisionedThroughputException;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseManager;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseTaker;
import com.amazonaws.services.kinesis.metrics.impl.MetricsHelper;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsScope;
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel;
import software.amazon.kinesis.leases.exceptions.DependencyException;
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
import software.amazon.kinesis.leases.exceptions.ProvisionedThroughputException;
import software.amazon.kinesis.metrics.MetricsHelper;
import software.amazon.kinesis.metrics.IMetricsScope;
import software.amazon.kinesis.metrics.MetricsLevel;
import lombok.extern.slf4j.Slf4j;

2
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/NoOpShardPrioritization.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/NoOpShardPrioritization.java
View file

@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.leases;
import java.util.List;

2
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ParentsFirstShardPrioritization.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ParentsFirstShardPrioritization.java
View file

@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.leases;
import java.util.ArrayList;
import java.util.Collections;

8
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardInfo.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ShardInfo.java
View file

@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.leases;
import java.util.Collection;
import java.util.Collections;
@ -22,7 +22,7 @@ import java.util.List;
import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
/**
* Used to pass shard related info among different classes and as a key to the map of shard consumers.
@ -86,7 +86,7 @@ public class ShardInfo {
*
* @return a list of shardId's that are parents of this shard, or empty if the shard has no parents.
*/
protected List<String> getParentShardIds() {
public List<String> getParentShardIds() {
return new LinkedList<String>(parentShardIds);
}
@ -95,7 +95,7 @@ public class ShardInfo {
*
* @return completion status of the shard
*/
protected boolean isCompleted() {
public boolean isCompleted() {
return ExtendedSequenceNumber.SHARD_END.equals(checkpoint);
}

2
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardPrioritization.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ShardPrioritization.java
View file

@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.leases;
import java.util.List;

16
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTask.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ShardSyncTask.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,11 +12,13 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.leases;
import com.amazonaws.services.kinesis.clientlibrary.proxies.IKinesisProxy;
import com.amazonaws.services.kinesis.leases.impl.KinesisClientLease;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseManager;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStreamExtended;
import software.amazon.kinesis.lifecycle.ITask;
import software.amazon.kinesis.lifecycle.TaskResult;
import software.amazon.kinesis.lifecycle.TaskType;
import software.amazon.kinesis.retrieval.IKinesisProxy;
import lombok.extern.slf4j.Slf4j;
@ -27,7 +29,7 @@ import lombok.extern.slf4j.Slf4j;
* cleanupLeasesUponShardCompletion is true).
*/
@Slf4j
class ShardSyncTask implements ITask {
public class ShardSyncTask implements ITask {
private final IKinesisProxy kinesisProxy;
private final ILeaseManager<KinesisClientLease> leaseManager;
private InitialPositionInStreamExtended initialPosition;
@ -43,7 +45,7 @@ class ShardSyncTask implements ITask {
* start processing records from this point in the stream (when an application starts up for the first time)
* except for shards that already have a checkpoint (and their descendant shards).
*/
ShardSyncTask(IKinesisProxy kinesisProxy,
public ShardSyncTask(IKinesisProxy kinesisProxy,
ILeaseManager<KinesisClientLease> leaseManager,
InitialPositionInStreamExtended initialPositionInStream,
boolean cleanupLeasesUponShardCompletion,

20
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTaskManager.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ShardSyncTaskManager.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,17 +12,19 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.leases;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import com.amazonaws.services.kinesis.clientlibrary.proxies.IKinesisProxy;
import com.amazonaws.services.kinesis.leases.impl.KinesisClientLease;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseManager;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStreamExtended;
import software.amazon.kinesis.lifecycle.ITask;
import software.amazon.kinesis.lifecycle.TaskResult;
import software.amazon.kinesis.metrics.MetricsCollectingTaskDecorator;
import software.amazon.kinesis.retrieval.IKinesisProxy;
import software.amazon.kinesis.metrics.IMetricsFactory;
import lombok.extern.slf4j.Slf4j;
@ -32,7 +34,7 @@ import lombok.extern.slf4j.Slf4j;
* Worker will use this class to kick off a sync task when it finds shards which have been completely processed.
*/
@Slf4j
class ShardSyncTaskManager {
public class ShardSyncTaskManager {
private ITask currentTask;
private Future<TaskResult> future;
private final IKinesisProxy kinesisProxy;
@ -58,7 +60,7 @@ class ShardSyncTaskManager {
* @param metricsFactory Metrics factory
* @param executorService ExecutorService to execute the shard sync tasks
*/
ShardSyncTaskManager(final IKinesisProxy kinesisProxy,
public ShardSyncTaskManager(final IKinesisProxy kinesisProxy,
final ILeaseManager<KinesisClientLease> leaseManager,
final InitialPositionInStreamExtended initialPositionInStream,
final boolean cleanupLeasesUponShardCompletion,
@ -76,7 +78,7 @@ class ShardSyncTaskManager {
this.initialPositionInStream = initialPositionInStream;
}
synchronized boolean syncShardAndLeaseInfo(Set<String> closedShardIds) {
public synchronized boolean syncShardAndLeaseInfo(Set<String> closedShardIds) {
return checkAndSubmitNextTask(closedShardIds);
}

30
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncer.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ShardSyncer.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.leases;
import java.io.Serializable;
import java.math.BigInteger;
@ -26,18 +26,20 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStreamExtended;
import org.apache.commons.lang.StringUtils;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.internal.KinesisClientLibIOException;
import com.amazonaws.services.kinesis.clientlibrary.proxies.IKinesisProxy;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import com.amazonaws.services.kinesis.leases.exceptions.DependencyException;
import com.amazonaws.services.kinesis.leases.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.leases.exceptions.ProvisionedThroughputException;
import com.amazonaws.services.kinesis.leases.impl.KinesisClientLease;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseManager;
import com.amazonaws.services.kinesis.metrics.impl.MetricsHelper;
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel;
import software.amazon.kinesis.retrieval.IKinesisProxy;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import software.amazon.kinesis.leases.exceptions.DependencyException;
import software.amazon.kinesis.leases.exceptions.InvalidStateException;
import software.amazon.kinesis.leases.exceptions.ProvisionedThroughputException;
import software.amazon.kinesis.leases.KinesisClientLease;
import software.amazon.kinesis.leases.ILeaseManager;
import software.amazon.kinesis.metrics.MetricsHelper;
import software.amazon.kinesis.metrics.MetricsLevel;
import com.amazonaws.services.kinesis.model.Shard;
import lombok.extern.slf4j.Slf4j;
@ -49,7 +51,7 @@ import lombok.extern.slf4j.Slf4j;
* and begun processing it's child shards.
*/
@Slf4j
class ShardSyncer {
public class ShardSyncer {
/**
* Note constructor is private: We use static synchronized methods - this is a utility class.
@ -80,7 +82,7 @@ class ShardSyncer {
* @throws ProvisionedThroughputException
* @throws KinesisClientLibIOException
*/
static synchronized void checkAndCreateLeasesForNewShards(IKinesisProxy kinesisProxy,
public static synchronized void checkAndCreateLeasesForNewShards(IKinesisProxy kinesisProxy,
ILeaseManager<KinesisClientLease> leaseManager,
InitialPositionInStreamExtended initialPositionInStream,
boolean cleanupLeasesOfCompletedShards,
@ -771,7 +773,7 @@ class ShardSyncer {
* @param shard
* @return
*/
static KinesisClientLease newKCLLease(Shard shard) {
public static KinesisClientLease newKCLLease(Shard shard) {
KinesisClientLease newLease = new KinesisClientLease();
newLease.setLeaseKey(shard.getShardId());
List<String> parentShardIds = new ArrayList<String>(2);

34
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/exceptions/DependencyException.java Normal file
View file

@ -0,0 +1,34 @@
/*
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 software.amazon.kinesis.leases.exceptions;
/**
* Indicates that a lease operation has failed because a dependency of the leasing system has failed. This will happen
* if DynamoDB throws an InternalServerException or a generic AmazonClientException (the specific subclasses of
* AmazonClientException are all handled more gracefully).
*/
public class DependencyException extends LeasingException {
private static final long serialVersionUID = 1L;
public DependencyException(Throwable e) {
super(e);
}
public DependencyException(String message, Throwable e) {
super(message, e);
}
}

37
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/exceptions/InvalidStateException.java Normal file
View file

@ -0,0 +1,37 @@
/*
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 software.amazon.kinesis.leases.exceptions;
/**
* Indicates that a lease operation has failed because DynamoDB is an invalid state. The most common example is failing
* to create the DynamoDB table before doing any lease operations.
*/
public class InvalidStateException extends LeasingException {
private static final long serialVersionUID = 1L;
public InvalidStateException(Throwable e) {
super(e);
}
public InvalidStateException(String message, Throwable e) {
super(message, e);
}
public InvalidStateException(String message) {
super(message);
}
}

36
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/exceptions/LeasingException.java Normal file
View file

@ -0,0 +1,36 @@
/*
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 software.amazon.kinesis.leases.exceptions;
/**
* Top-level exception type for all exceptions thrown by the leasing code.
*/
public class LeasingException extends Exception {
public LeasingException(Throwable e) {
super(e);
}
public LeasingException(String message, Throwable e) {
super(message, e);
}
public LeasingException(String message) {
super(message);
}
private static final long serialVersionUID = 1L;
}

32
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/exceptions/ProvisionedThroughputException.java Normal file
View file

@ -0,0 +1,32 @@
/*
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 software.amazon.kinesis.leases.exceptions;
/**
* Indicates that a lease operation has failed due to lack of provisioned throughput for a DynamoDB table.
*/
public class ProvisionedThroughputException extends LeasingException {
private static final long serialVersionUID = 1L;
public ProvisionedThroughputException(Throwable e) {
super(e);
}
public ProvisionedThroughputException(String message, Throwable e) {
super(message, e);
}
}

13
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/BlockOnParentShardTask.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/BlockOnParentShardTask.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,12 +12,13 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.lifecycle;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.internal.BlockedOnParentShardException;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import com.amazonaws.services.kinesis.leases.impl.KinesisClientLease;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseManager;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import software.amazon.kinesis.leases.KinesisClientLease;
import software.amazon.kinesis.leases.ILeaseManager;
import lombok.extern.slf4j.Slf4j;
@ -30,7 +31,7 @@ import lombok.extern.slf4j.Slf4j;
* proceed with processing data from the shard.
*/
@Slf4j
class BlockOnParentShardTask implements ITask {
public class BlockOnParentShardTask implements ITask {
private final ShardInfo shardInfo;
private final ILeaseManager<KinesisClientLease> leaseManager;

2
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ConsumerStates.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ConsumerStates.java
View file

@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.lifecycle;
/**
* Top level container for all the possible states a {@link ShardConsumer} can be in. The logic for creation of tasks,

41
amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ITask.java Normal file
View file

@ -0,0 +1,41 @@
/*
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 software.amazon.kinesis.lifecycle;
import software.amazon.kinesis.lifecycle.TaskResult;
import software.amazon.kinesis.lifecycle.TaskType;
import java.util.concurrent.Callable;
/**
* Interface for shard processing tasks.
* A task may execute an application callback (e.g. initialize, process, shutdown).
*/
public interface ITask extends Callable<TaskResult> {
/**
* Perform task logic.
* E.g. perform set up (e.g. fetch records) and invoke a callback (e.g. processRecords() API).
*
* @return TaskResult (captures any exceptions encountered during execution of the task)
*/
TaskResult call();
/**
* @return TaskType
*/
TaskType getTaskType();
}

7
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/types/InitializationInput.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/InitializationInput.java
View file

@ -12,11 +12,14 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.types;
package software.amazon.kinesis.lifecycle;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
/**
* Container for the parameters to the IRecordProcessor's
* {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor#initialize(InitializationInput
* {@link IRecordProcessor#initialize(InitializationInput
* initializationInput) initialize} method.
*/
public class InitializationInput {

22
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/InitializeTask.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/InitializeTask.java
View file

@ -12,15 +12,19 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.lifecycle;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.ICheckpoint;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
import com.amazonaws.services.kinesis.clientlibrary.lib.checkpoint.Checkpoint;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import com.amazonaws.services.kinesis.clientlibrary.types.InitializationInput;
import com.amazonaws.services.kinesis.metrics.impl.MetricsHelper;
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel;
import software.amazon.kinesis.coordinator.RecordProcessorCheckpointer;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.coordinator.StreamConfig;
import software.amazon.kinesis.processor.ICheckpoint;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
import software.amazon.kinesis.checkpoint.Checkpoint;
import software.amazon.kinesis.retrieval.GetRecordsCache;
import software.amazon.kinesis.retrieval.KinesisDataFetcher;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import software.amazon.kinesis.metrics.MetricsHelper;
import software.amazon.kinesis.metrics.MetricsLevel;
import lombok.extern.slf4j.Slf4j;
@ -28,7 +32,7 @@ import lombok.extern.slf4j.Slf4j;
* Task for initializing shard position and invoking the RecordProcessor initialize() API.
*/
@Slf4j
class InitializeTask implements ITask {
public class InitializeTask implements ITask {
private static final String RECORD_PROCESSOR_INITIALIZE_METRIC = "RecordProcessor.initialize";
private final ShardInfo shardInfo;

7
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/types/ProcessRecordsInput.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ProcessRecordsInput.java
View file

@ -12,20 +12,21 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.types;
package software.amazon.kinesis.lifecycle;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
import software.amazon.kinesis.processor.IRecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.model.Record;
import lombok.Getter;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
/**
* Container for the parameters to the IRecordProcessor's
* {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor#processRecords(
* {@link IRecordProcessor#processRecords(
* ProcessRecordsInput processRecordsInput) processRecords} method.
*/
public class ProcessRecordsInput {

27
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ProcessTask.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ProcessTask.java
View file

@ -12,22 +12,27 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.lifecycle;
import java.math.BigInteger;
import java.util.List;
import java.util.ListIterator;
import com.amazonaws.services.cloudwatch.model.StandardUnit;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
import com.amazonaws.services.kinesis.clientlibrary.proxies.IKinesisProxy;
import com.amazonaws.services.kinesis.clientlibrary.proxies.IKinesisProxyExtended;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import com.amazonaws.services.kinesis.clientlibrary.types.ProcessRecordsInput;
import com.amazonaws.services.kinesis.clientlibrary.types.UserRecord;
import com.amazonaws.services.kinesis.metrics.impl.MetricsHelper;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsScope;
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel;
import software.amazon.kinesis.coordinator.RecordProcessorCheckpointer;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.coordinator.StreamConfig;
import software.amazon.kinesis.retrieval.ThrottlingReporter;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
import software.amazon.kinesis.retrieval.GetRecordsCache;
import software.amazon.kinesis.retrieval.IKinesisProxy;
import software.amazon.kinesis.retrieval.IKinesisProxyExtended;
import software.amazon.kinesis.retrieval.KinesisDataFetcher;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import software.amazon.kinesis.retrieval.kpl.UserRecord;
import software.amazon.kinesis.metrics.MetricsHelper;
import software.amazon.kinesis.metrics.IMetricsScope;
import software.amazon.kinesis.metrics.MetricsLevel;
import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
import com.amazonaws.services.kinesis.model.Record;
@ -39,7 +44,7 @@ import lombok.extern.slf4j.Slf4j;
* Task for fetching data records and invoking processRecords() on the record processor instance.
*/
@Slf4j
class ProcessTask implements ITask {
public class ProcessTask implements ITask {
private static final String EXPIRED_ITERATOR_METRIC = "ExpiredIterator";
private static final String DATA_BYTES_PROCESSED_METRIC = "DataBytesProcessed";
private static final String RECORDS_PROCESSED_METRIC = "RecordsProcessed";

41
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumer.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShardConsumer.java
View file

@ -12,7 +12,7 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.lifecycle;
import java.util.Optional;
@ -21,15 +21,26 @@ import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.internal.BlockedOnParentShardException;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.ICheckpoint;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
import com.amazonaws.services.kinesis.leases.impl.KinesisClientLease;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseManager;
import com.amazonaws.services.kinesis.metrics.interfaces.IMetricsFactory;
import software.amazon.kinesis.coordinator.KinesisClientLibConfiguration;
import software.amazon.kinesis.checkpoint.Checkpoint;
import software.amazon.kinesis.metrics.MetricsCollectingTaskDecorator;
import software.amazon.kinesis.coordinator.RecordProcessorCheckpointer;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.coordinator.StreamConfig;
import software.amazon.kinesis.processor.ICheckpoint;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
import software.amazon.kinesis.leases.KinesisClientLease;
import software.amazon.kinesis.leases.ILeaseManager;
import software.amazon.kinesis.metrics.IMetricsFactory;
import com.google.common.annotations.VisibleForTesting;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import software.amazon.kinesis.retrieval.AsynchronousGetRecordsRetrievalStrategy;
import software.amazon.kinesis.retrieval.GetRecordsCache;
import software.amazon.kinesis.retrieval.GetRecordsRetrievalStrategy;
import software.amazon.kinesis.retrieval.KinesisDataFetcher;
import software.amazon.kinesis.retrieval.SynchronousGetRecordsRetrievalStrategy;
/**
* Responsible for consuming data records of a (specified) shard.
@ -37,7 +48,7 @@ import lombok.extern.slf4j.Slf4j;
* A new instance should be created if the primary responsibility is reassigned back to this process.
*/
@Slf4j
class ShardConsumer {
public class ShardConsumer {
private final StreamConfig streamConfig;
private final IRecordProcessor recordProcessor;
private final KinesisClientLibConfiguration config;
@ -140,7 +151,7 @@ class ShardConsumer {
* @param config Kinesis library configuration
*/
// CHECKSTYLE:IGNORE ParameterNumber FOR NEXT 10 LINES
ShardConsumer(ShardInfo shardInfo,
public ShardConsumer(ShardInfo shardInfo,
StreamConfig streamConfig,
ICheckpoint checkpoint,
IRecordProcessor recordProcessor,
@ -163,7 +174,7 @@ class ShardConsumer {
new RecordProcessorCheckpointer(
shardInfo,
checkpoint,
new SequenceNumberValidator(
new Checkpoint.SequenceNumberValidator(
streamConfig.getStreamProxy(),
shardInfo.getShardId(),
streamConfig.shouldValidateSequenceNumberBeforeCheckpointing()),
@ -241,7 +252,7 @@ class ShardConsumer {
*
* @return true if a new process task was submitted, false otherwise
*/
synchronized boolean consumeShard() {
public synchronized boolean consumeShard() {
return checkAndSubmitNextTask();
}
@ -340,7 +351,7 @@ class ShardConsumer {
*
* @param shutdownNotification used to signal that the record processor has been given the chance to shutdown.
*/
void notifyShutdownRequested(ShutdownNotification shutdownNotification) {
public void notifyShutdownRequested(ShutdownNotification shutdownNotification) {
this.shutdownNotification = shutdownNotification;
markForShutdown(ShutdownReason.REQUESTED);
}
@ -351,7 +362,7 @@ class ShardConsumer {
*
* @return true if shutdown is complete (false if shutdown is still in progress)
*/
synchronized boolean beginShutdown() {
public synchronized boolean beginShutdown() {
markForShutdown(ShutdownReason.ZOMBIE);
checkAndSubmitNextTask();
@ -371,14 +382,14 @@ class ShardConsumer {
*
* @return true if shutdown is complete
*/
boolean isShutdown() {
public boolean isShutdown() {
return currentState.isTerminal();
}
/**
* @return the shutdownReason
*/
ShutdownReason getShutdownReason() {
public ShutdownReason getShutdownReason() {
return shutdownReason;
}
@ -425,7 +436,7 @@ class ShardConsumer {
}
@VisibleForTesting
boolean isShutdownRequested() {
public boolean isShutdownRequested() {
return shutdownReason != null;
}

12
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumerShutdownNotification.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShardConsumerShutdownNotification.java
View file

@ -12,19 +12,19 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.lifecycle;
import java.util.concurrent.CountDownLatch;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware;
import com.amazonaws.services.kinesis.leases.impl.KinesisClientLease;
import com.amazonaws.services.kinesis.leases.impl.LeaseCoordinator;
import software.amazon.kinesis.processor.v2.IShutdownNotificationAware;
import software.amazon.kinesis.leases.KinesisClientLease;
import software.amazon.kinesis.leases.LeaseCoordinator;
/**
* Contains callbacks for completion of stages in a requested record processor shutdown.
*
*/
class ShardConsumerShutdownNotification implements ShutdownNotification {
public class ShardConsumerShutdownNotification implements ShutdownNotification {
private final LeaseCoordinator<KinesisClientLease> leaseCoordinator;
private final KinesisClientLease lease;
@ -48,7 +48,7 @@ class ShardConsumerShutdownNotification implements ShutdownNotification {
* @param shutdownCompleteLatch
* used to inform the caller once the record processor is fully shutdown
*/
ShardConsumerShutdownNotification(LeaseCoordinator<KinesisClientLease> leaseCoordinator, KinesisClientLease lease,
public ShardConsumerShutdownNotification(LeaseCoordinator<KinesisClientLease> leaseCoordinator, KinesisClientLease lease,
CountDownLatch notificationCompleteLatch, CountDownLatch shutdownCompleteLatch) {
this.leaseCoordinator = leaseCoordinator;
this.lease = lease;

8
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/types/ShutdownInput.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShutdownInput.java
View file

@ -12,14 +12,14 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.types;
package software.amazon.kinesis.lifecycle;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;
import software.amazon.kinesis.processor.IRecordProcessorCheckpointer;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
/**
* Container for the parameters to the IRecordProcessor's
* {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor#shutdown(ShutdownInput
* {@link IRecordProcessor#shutdown(ShutdownInput
* shutdownInput) shutdown} method.
*/
public class ShutdownInput {

7
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShutdownNotification.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShutdownNotification.java
View file

@ -12,9 +12,10 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.lifecycle;
import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownInput;
import software.amazon.kinesis.lifecycle.ShutdownInput;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
/**
* A shutdown request to the ShardConsumer
@ -29,7 +30,7 @@ public interface ShutdownNotification {
/**
* Used to indicate that the record processor has completed the call to
* {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor#shutdown(ShutdownInput)} has
* {@link IRecordProcessor#shutdown(ShutdownInput)} has
* completed.
*/
void shutdownComplete();

11
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShutdownNotificationTask.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShutdownNotificationTask.java
View file

@ -12,16 +12,17 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.lifecycle;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IShutdownNotificationAware;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.processor.IRecordProcessorCheckpointer;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
import software.amazon.kinesis.processor.v2.IShutdownNotificationAware;
/**
* Notifies record processor of incoming shutdown request, and gives them a chance to checkpoint.
*/
class ShutdownNotificationTask implements ITask {
public class ShutdownNotificationTask implements ITask {
private final IRecordProcessor recordProcessor;
private final IRecordProcessorCheckpointer recordProcessorCheckpointer;

11
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShutdownReason.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShutdownReason.java
View file

@ -12,11 +12,12 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.lifecycle;
import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownInput;
import static com.amazonaws.services.kinesis.clientlibrary.lib.worker.ConsumerStates.ConsumerState;
import static com.amazonaws.services.kinesis.clientlibrary.lib.worker.ConsumerStates.ShardConsumerState;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
import static software.amazon.kinesis.lifecycle.ConsumerStates.ConsumerState;
import static software.amazon.kinesis.lifecycle.ConsumerStates.ShardConsumerState;
/**
@ -46,7 +47,7 @@ public enum ShutdownReason {
/**
* Indicates that the entire application is being shutdown, and if desired the record processor will be given a
* final chance to checkpoint. This state will not trigger a direct call to
* {@link com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor#shutdown(ShutdownInput)}, but
* {@link IRecordProcessor#shutdown(ShutdownInput)}, but
* instead depend on a different interface for backward compatibility.
*/
REQUESTED(1, ShardConsumerState.SHUTDOWN_REQUESTED.getConsumerState());

26
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShutdownTask.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShutdownTask.java
View file

@ -12,16 +12,20 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.lifecycle;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
import com.amazonaws.services.kinesis.clientlibrary.proxies.IKinesisProxy;
import com.amazonaws.services.kinesis.clientlibrary.types.ExtendedSequenceNumber;
import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownInput;
import com.amazonaws.services.kinesis.leases.impl.KinesisClientLease;
import com.amazonaws.services.kinesis.leases.interfaces.ILeaseManager;
import com.amazonaws.services.kinesis.metrics.impl.MetricsHelper;
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStreamExtended;
import software.amazon.kinesis.coordinator.RecordProcessorCheckpointer;
import software.amazon.kinesis.leases.ShardInfo;
import software.amazon.kinesis.leases.ShardSyncer;
import software.amazon.kinesis.processor.v2.IRecordProcessor;
import software.amazon.kinesis.retrieval.GetRecordsCache;
import software.amazon.kinesis.retrieval.IKinesisProxy;
import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
import software.amazon.kinesis.leases.KinesisClientLease;
import software.amazon.kinesis.leases.ILeaseManager;
import software.amazon.kinesis.metrics.MetricsHelper;
import software.amazon.kinesis.metrics.MetricsLevel;
import com.google.common.annotations.VisibleForTesting;
import lombok.extern.slf4j.Slf4j;
@ -30,7 +34,7 @@ import lombok.extern.slf4j.Slf4j;
* Task for invoking the RecordProcessor shutdown() callback.
*/
@Slf4j
class ShutdownTask implements ITask {
public class ShutdownTask implements ITask {
private static final String RECORD_PROCESSOR_SHUTDOWN_METRIC = "RecordProcessor.shutdown";
private final ShardInfo shardInfo;
@ -162,7 +166,7 @@ class ShutdownTask implements ITask {
}
@VisibleForTesting
ShutdownReason getReason() {
public ShutdownReason getReason() {
return reason;
}

10
amazon-kinesis-client/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/TaskResult.java → amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/TaskResult.java
View file

@ -1,5 +1,5 @@
/*
* Copyright 2012-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
* Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
@ -12,13 +12,13 @@
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
package com.amazonaws.services.kinesis.clientlibrary.lib.worker;
package software.amazon.kinesis.lifecycle;
/**
* Used to capture information from a task that we want to communicate back to the higher layer.
* E.g. exception thrown when executing the task, if we reach end of a shard.
*/
class TaskResult {
public class TaskResult {
// Did we reach the end of the shard while processing this task.
private boolean shardEndReached;
@ -29,7 +29,7 @@ class TaskResult {
/**
* @return the shardEndReached
*/
protected boolean isShardEndReached() {
public boolean isShardEndReached() {
return shardEndReached;
}
@ -50,7 +50,7 @@ class TaskResult {
/**
* @param e Any exception encountered when running the process task.
*/
TaskResult(Exception e) {
public TaskResult(Exception e) {
this(e, false);
}

Some files were not shown because too many files have changed in this diff Show more