From 2afb0bd00ba54b92626387586ebf7ac9f2b0e9bc Mon Sep 17 00:00:00 2001
From: Mike Watters <mike.watters@adroll.com>
Date: Wed, 13 Sep 2017 14:17:32 -0600
Subject: [PATCH] allow unexpected child shards to be ignored

now instead of always throwing an assertion if a child shard has an
open parent, consider worker configuration before doing so.  if
configured to ignore such shards, do not create leases for them during
shard sync.  this is intended to mitigate failing worker init when
processing dynamodb streams with many thousands of shards (which can
happen for tables with thousands of partitions).

this new behavior can be enabled by adding the following to a
configuration/properties file:

```
ignoreUnexpectedChildShards = true
```
---
 .../lib/worker/ConsumerStates.java            |  18 +-
 .../worker/KinesisClientLibConfiguration.java |  31 ++-
 .../lib/worker/ShardConsumer.java             |  26 +-
 .../lib/worker/ShardSyncTask.java             |   6 +-
 .../lib/worker/ShardSyncTaskManager.java      |   7 +-
 .../clientlibrary/lib/worker/ShardSyncer.java | 230 +++++++++++-------
 .../lib/worker/ShutdownTask.java              |  12 +-
 .../clientlibrary/lib/worker/Worker.java      |  39 +--
 .../KinesisClientLibConfigurationTest.java    |   4 +
 .../lib/worker/ShardConsumerTest.java         |   9 +
 .../worker/ShardSyncTaskIntegrationTest.java  |   4 +-
 .../lib/worker/ShutdownTaskTest.java          |   6 +-
 .../clientlibrary/lib/worker/WorkerTest.java  |  65 +++--
 13 files changed, 299 insertions(+), 158 deletions(-)

diff --git a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ConsumerStates.java b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ConsumerStates.java
index d3ccb911..f96d622b 100644
--- a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ConsumerStates.java
+++ b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ConsumerStates.java
@@ -21,7 +21,7 @@ import java.util.Optional;
  * and state transitions is contained within the {@link ConsumerState} objects.
  *
  * <h2>State Diagram</h2>
- * 
+ *
  * <pre>
  *       +-------------------+
  *       | Waiting on Parent |                               +------------------+
@@ -96,14 +96,14 @@ class ConsumerStates {
     /**
      * Represents a the current state of the consumer. This handles the creation of tasks for the consumer, and what to
      * do when a transition occurs.
-     * 
+     *
      */
     interface ConsumerState {
         /**
          * Creates a new task for this state using the passed in consumer to build the task. If there is no task
          * required for this state it may return a null value. {@link ConsumerState}'s are allowed to modify the
          * consumer during the execution of this method.
-         * 
+         *
          * @param consumer
          *            the consumer to use build the task, or execute state.
          * @return a valid task for this state or null if there is no task required.
@@ -113,7 +113,7 @@ class ConsumerStates {
         /**
          * Provides the next state of the consumer upon success of the task return by
          * {@link ConsumerState#createTask(ShardConsumer)}.
-         * 
+         *
          * @return the next state that the consumer should transition to, this may be the same object as the current
          *         state.
          */
@@ -122,7 +122,7 @@ class ConsumerStates {
         /**
          * Provides the next state of the consumer when a shutdown has been requested. The returned state is dependent
          * on the current state, and the shutdown reason.
-         * 
+         *
          * @param shutdownReason
          *            the reason that a shutdown was requested
          * @return the next state that the consumer should transition to, this may be the same object as the current
@@ -133,7 +133,7 @@ class ConsumerStates {
         /**
          * The type of task that {@link ConsumerState#createTask(ShardConsumer)} would return. This is always a valid state
          * even if createTask would return a null value.
-         * 
+         *
          * @return the type of task that this state represents.
          */
         TaskType getTaskType();
@@ -141,7 +141,7 @@ class ConsumerStates {
         /**
          * An enumeration represent the type of this state. Different consumer states may return the same
          * {@link ShardConsumerState}.
-         * 
+         *
          * @return the type of consumer state this represents.
          */
         ShardConsumerState getState();
@@ -515,7 +515,9 @@ class ConsumerStates {
                     consumer.getRecordProcessorCheckpointer(), consumer.getShutdownReason(),
                     consumer.getStreamConfig().getStreamProxy(),
                     consumer.getStreamConfig().getInitialPositionInStream(),
-                    consumer.isCleanupLeasesOfCompletedShards(), consumer.getLeaseManager(),
+                    consumer.isCleanupLeasesOfCompletedShards(),
+                    consumer.isIgnoreUnexpectedChildShards(),
+                    consumer.getLeaseManager(),
                     consumer.getTaskBackoffTimeMillis(),
                     consumer.getGetRecordsRetrievalStrategy());
         }
diff --git a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java
index 7d6dac5a..8779f47a 100644
--- a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java
+++ b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java
@@ -86,6 +86,11 @@ public class KinesisClientLibConfiguration {
      */
     public static final boolean DEFAULT_CLEANUP_LEASES_UPON_SHARDS_COMPLETION = true;
 
+    /**
+     * Ignore child shards with open parents.
+     */
+    public static final boolean DEFAULT_IGNORE_UNEXPECTED_CHILD_SHARDS = false;
+
     /**
      * Backoff time in milliseconds for Amazon Kinesis Client Library tasks (in the event of failures).
      */
@@ -200,6 +205,7 @@ public class KinesisClientLibConfiguration {
     private boolean callProcessRecordsEvenForEmptyRecordList;
     private long parentShardPollIntervalMillis;
     private boolean cleanupLeasesUponShardCompletion;
+    private boolean ignoreUnexpectedChildShards;
     private ClientConfiguration kinesisClientConfig;
     private ClientConfiguration dynamoDBClientConfig;
     private ClientConfiguration cloudWatchClientConfig;
@@ -272,6 +278,7 @@ public class KinesisClientLibConfiguration {
                 DEFAULT_MAX_RECORDS, DEFAULT_IDLETIME_BETWEEN_READS_MILLIS,
                 DEFAULT_DONT_CALL_PROCESS_RECORDS_FOR_EMPTY_RECORD_LIST, DEFAULT_PARENT_SHARD_POLL_INTERVAL_MILLIS,
                 DEFAULT_SHARD_SYNC_INTERVAL_MILLIS, DEFAULT_CLEANUP_LEASES_UPON_SHARDS_COMPLETION,
+                DEFAULT_IGNORE_UNEXPECTED_CHILD_SHARDS,
                 new ClientConfiguration(), new ClientConfiguration(), new ClientConfiguration(),
                 DEFAULT_TASK_BACKOFF_TIME_MILLIS, DEFAULT_METRICS_BUFFER_TIME_MILLIS, DEFAULT_METRICS_MAX_QUEUE_SIZE,
                 DEFAULT_VALIDATE_SEQUENCE_NUMBER_BEFORE_CHECKPOINTING, null,
@@ -300,6 +307,7 @@ public class KinesisClientLibConfiguration {
      * @param shardSyncIntervalMillis Time between tasks to sync leases and Kinesis shards
      * @param cleanupTerminatedShardsBeforeExpiry Clean up shards we've finished processing (don't wait for expiration
      *        in Kinesis)
+     * @param ignoreUnexpectedChildShards Ignore child shards with open parents
      * @param kinesisClientConfig Client Configuration used by Kinesis client
      * @param dynamoDBClientConfig Client Configuration used by DynamoDB client
      * @param cloudWatchClientConfig Client Configuration used by CloudWatch client
@@ -329,6 +337,7 @@ public class KinesisClientLibConfiguration {
             long parentShardPollIntervalMillis,
             long shardSyncIntervalMillis,
             boolean cleanupTerminatedShardsBeforeExpiry,
+            boolean ignoreUnexpectedChildShards,
             ClientConfiguration kinesisClientConfig,
             ClientConfiguration dynamoDBClientConfig,
             ClientConfiguration cloudWatchClientConfig,
@@ -342,7 +351,7 @@ public class KinesisClientLibConfiguration {
                 dynamoDBCredentialsProvider, cloudWatchCredentialsProvider, failoverTimeMillis, workerId,
                 maxRecords, idleTimeBetweenReadsInMillis,
                 callProcessRecordsEvenForEmptyRecordList, parentShardPollIntervalMillis,
-                shardSyncIntervalMillis, cleanupTerminatedShardsBeforeExpiry,
+                shardSyncIntervalMillis, cleanupTerminatedShardsBeforeExpiry, ignoreUnexpectedChildShards,
                 kinesisClientConfig, dynamoDBClientConfig, cloudWatchClientConfig,
                 taskBackoffTimeMillis, metricsBufferTimeMillis, metricsMaxQueueSize,
              validateSequenceNumberBeforeCheckpointing, regionName, shutdownGraceMillis);
@@ -371,6 +380,7 @@ public class KinesisClientLibConfiguration {
      * @param shardSyncIntervalMillis Time between tasks to sync leases and Kinesis shards
      * @param cleanupTerminatedShardsBeforeExpiry Clean up shards we've finished processing (don't wait for expiration
      *        in Kinesis)
+     * @param ignoreUnexpectedChildShards Ignore child shards with open parents
      * @param kinesisClientConfig Client Configuration used by Kinesis client
      * @param dynamoDBClientConfig Client Configuration used by DynamoDB client
      * @param cloudWatchClientConfig Client Configuration used by CloudWatch client
@@ -400,6 +410,7 @@ public class KinesisClientLibConfiguration {
             long parentShardPollIntervalMillis,
             long shardSyncIntervalMillis,
             boolean cleanupTerminatedShardsBeforeExpiry,
+            boolean ignoreUnexpectedChildShards,
             ClientConfiguration kinesisClientConfig,
             ClientConfiguration dynamoDBClientConfig,
             ClientConfiguration cloudWatchClientConfig,
@@ -436,6 +447,7 @@ public class KinesisClientLibConfiguration {
         this.parentShardPollIntervalMillis = parentShardPollIntervalMillis;
         this.shardSyncIntervalMillis = shardSyncIntervalMillis;
         this.cleanupLeasesUponShardCompletion = cleanupTerminatedShardsBeforeExpiry;
+        this.ignoreUnexpectedChildShards = ignoreUnexpectedChildShards;
         this.workerIdentifier = workerId;
         this.kinesisClientConfig = checkAndAppendKinesisClientLibUserAgent(kinesisClientConfig);
         this.dynamoDBClientConfig = checkAndAppendKinesisClientLibUserAgent(dynamoDBClientConfig);
@@ -670,6 +682,13 @@ public class KinesisClientLibConfiguration {
         return cleanupLeasesUponShardCompletion;
     }
 
+    /**
+     * @return true if we should ignore child shards which have open parents
+     */
+    public boolean shouldIgnoreUnexpectedChildShards() {
+        return ignoreUnexpectedChildShards;
+    }
+
     /**
      * @return true if KCL should validate client provided sequence numbers with a call to Amazon Kinesis before
      *         checkpointing for calls to {@link RecordProcessorCheckpointer#checkpoint(String)}
@@ -890,6 +909,16 @@ public class KinesisClientLibConfiguration {
         return this;
     }
 
+    /**
+     * @param ignoreUnexpectedChildShards Ignore child shards with open parents.
+     * @return KinesisClientLibConfiguration
+     */
+    public KinesisClientLibConfiguration withIgnoreUnexpectedChildShards(
+            boolean ignoreUnexpectedChildShards) {
+        this.ignoreUnexpectedChildShards = ignoreUnexpectedChildShards;
+        return this;
+    }
+
     /**
      * @param clientConfig Common client configuration used by Kinesis/DynamoDB/CloudWatch client
      * @return KinesisClientLibConfiguration
diff --git a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumer.java b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumer.java
index 4bbe1939..246c2aa9 100644
--- a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumer.java
+++ b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumer.java
@@ -53,6 +53,7 @@ class ShardConsumer {
     // Backoff time when polling to check if application has finished processing parent shards
     private final long parentShardPollIntervalMillis;
     private final boolean cleanupLeasesOfCompletedShards;
+    private final boolean ignoreUnexpectedChildShards;
     private final long taskBackoffTimeMillis;
     private final boolean skipShardSyncAtWorkerInitializationIfLeasesExist;
 
@@ -97,7 +98,7 @@ class ShardConsumer {
      * @param metricsFactory IMetricsFactory used to construct IMetricsScopes for this shard
      * @param backoffTimeMillis backoff interval when we encounter exceptions
      */
-    // CHECKSTYLE:IGNORE ParameterNumber FOR NEXT 10 LINES
+    // CHECKSTYLE:IGNORE ParameterNumber FOR NEXT 11 LINES
     ShardConsumer(ShardInfo shardInfo,
             StreamConfig streamConfig,
             ICheckpoint checkpoint,
@@ -105,12 +106,13 @@ class ShardConsumer {
             ILeaseManager<KinesisClientLease> leaseManager,
             long parentShardPollIntervalMillis,
             boolean cleanupLeasesOfCompletedShards,
+            boolean ignoreUnexpectedChildShards,
             ExecutorService executorService,
             IMetricsFactory metricsFactory,
             long backoffTimeMillis,
             boolean skipShardSyncAtWorkerInitializationIfLeasesExist) {
         this(shardInfo, streamConfig, checkpoint,recordProcessor, leaseManager, parentShardPollIntervalMillis,
-                cleanupLeasesOfCompletedShards, executorService, metricsFactory, backoffTimeMillis,
+                cleanupLeasesOfCompletedShards, ignoreUnexpectedChildShards, executorService, metricsFactory, backoffTimeMillis,
                 skipShardSyncAtWorkerInitializationIfLeasesExist, Optional.empty(), Optional.empty());
     }
 
@@ -135,6 +137,7 @@ class ShardConsumer {
                   ILeaseManager<KinesisClientLease> leaseManager,
                   long parentShardPollIntervalMillis,
                   boolean cleanupLeasesOfCompletedShards,
+                  boolean ignoreUnexpectedChildShards,
                   ExecutorService executorService,
                   IMetricsFactory metricsFactory,
                   long backoffTimeMillis,
@@ -157,6 +160,7 @@ class ShardConsumer {
         this.metricsFactory = metricsFactory;
         this.parentShardPollIntervalMillis = parentShardPollIntervalMillis;
         this.cleanupLeasesOfCompletedShards = cleanupLeasesOfCompletedShards;
+        this.ignoreUnexpectedChildShards = ignoreUnexpectedChildShards;
         this.taskBackoffTimeMillis = backoffTimeMillis;
         this.skipShardSyncAtWorkerInitializationIfLeasesExist = skipShardSyncAtWorkerInitializationIfLeasesExist;
         this.getRecordsRetrievalStrategy = makeStrategy(dataFetcher, retryGetRecordsInSeconds, maxGetRecordsThreadPool, shardInfo);
@@ -165,7 +169,7 @@ class ShardConsumer {
     /**
      * No-op if current task is pending, otherwise submits next task for this shard.
      * This method should NOT be called if the ShardConsumer is already in SHUTDOWN_COMPLETED state.
-     * 
+     *
      * @return true if a new process task was submitted, false otherwise
      */
     synchronized boolean consumeShard() {
@@ -260,7 +264,7 @@ class ShardConsumer {
     /**
      * Requests the shutdown of the this ShardConsumer. This should give the record processor a chance to checkpoint
      * before being shutdown.
-     * 
+     *
      * @param shutdownNotification used to signal that the record processor has been given the chance to shutdown.
      */
     void notifyShutdownRequested(ShutdownNotification shutdownNotification) {
@@ -271,7 +275,7 @@ class ShardConsumer {
     /**
      * Shutdown this ShardConsumer (including invoking the RecordProcessor shutdown API).
      * This is called by Worker when it loses responsibility for a shard.
-     * 
+     *
      * @return true if shutdown is complete (false if shutdown is still in progress)
      */
     synchronized boolean beginShutdown() {
@@ -291,7 +295,7 @@ class ShardConsumer {
     /**
      * Used (by Worker) to check if this ShardConsumer instance has been shutdown
      * RecordProcessor shutdown() has been invoked, as appropriate.
-     * 
+     *
      * @return true if shutdown is complete
      */
     boolean isShutdown() {
@@ -307,7 +311,7 @@ class ShardConsumer {
 
     /**
      * Figure out next task to run based on current state, task, and shutdown context.
-     * 
+     *
      * @return Return next task to run
      */
     private ITask getNextTask() {
@@ -323,7 +327,7 @@ class ShardConsumer {
     /**
      * Note: This is a private/internal method with package level access solely for testing purposes.
      * Update state based on information about: task success, current state, and shutdown info.
-     * 
+     *
      * @param taskOutcome The outcome of the last task
      */
     void updateState(TaskOutcome taskOutcome) {
@@ -355,7 +359,7 @@ class ShardConsumer {
 
     /**
      * Private/Internal method - has package level access solely for testing purposes.
-     * 
+     *
      * @return the currentState
      */
     ConsumerStates.ShardConsumerState getCurrentState() {
@@ -402,6 +406,10 @@ class ShardConsumer {
         return cleanupLeasesOfCompletedShards;
     }
 
+    boolean isIgnoreUnexpectedChildShards() {
+        return ignoreUnexpectedChildShards;
+    }
+
     long getTaskBackoffTimeMillis() {
         return taskBackoffTimeMillis;
     }
diff --git a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTask.java b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTask.java
index ddfb8459..5a0c3d5a 100644
--- a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTask.java
+++ b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTask.java
@@ -35,6 +35,7 @@ class ShardSyncTask implements ITask {
     private final ILeaseManager<KinesisClientLease> leaseManager;
     private InitialPositionInStreamExtended initialPosition;
     private final boolean cleanupLeasesUponShardCompletion;
+    private final boolean ignoreUnexpectedChildShards;
     private final long shardSyncTaskIdleTimeMillis;
     private final TaskType taskType = TaskType.SHARDSYNC;
 
@@ -49,11 +50,13 @@ class ShardSyncTask implements ITask {
             ILeaseManager<KinesisClientLease> leaseManager,
             InitialPositionInStreamExtended initialPositionInStream,
             boolean cleanupLeasesUponShardCompletion,
+            boolean ignoreUnexpectedChildShards,
             long shardSyncTaskIdleTimeMillis) {
         this.kinesisProxy = kinesisProxy;
         this.leaseManager = leaseManager;
         this.initialPosition = initialPositionInStream;
         this.cleanupLeasesUponShardCompletion = cleanupLeasesUponShardCompletion;
+        this.ignoreUnexpectedChildShards = ignoreUnexpectedChildShards;
         this.shardSyncTaskIdleTimeMillis = shardSyncTaskIdleTimeMillis;
     }
 
@@ -68,7 +71,8 @@ class ShardSyncTask implements ITask {
             ShardSyncer.checkAndCreateLeasesForNewShards(kinesisProxy,
                     leaseManager,
                     initialPosition,
-                    cleanupLeasesUponShardCompletion);
+                    cleanupLeasesUponShardCompletion,
+                    ignoreUnexpectedChildShards);
             if (shardSyncTaskIdleTimeMillis > 0) {
                 Thread.sleep(shardSyncTaskIdleTimeMillis);
             }
diff --git a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTaskManager.java b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTaskManager.java
index c1bfae76..65b7fad6 100644
--- a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTaskManager.java
+++ b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTaskManager.java
@@ -44,17 +44,19 @@ class ShardSyncTaskManager {
     private final ExecutorService executorService;
     private final InitialPositionInStreamExtended initialPositionInStream;
     private boolean cleanupLeasesUponShardCompletion;
+    private boolean ignoreUnexpectedChildShards;
     private final long shardSyncIdleTimeMillis;
 
 
     /**
      * Constructor.
-     * 
+     *
      * @param kinesisProxy Proxy used to fetch streamInfo (shards)
      * @param leaseManager Lease manager (used to list and create leases for shards)
      * @param initialPositionInStream Initial position in stream
      * @param cleanupLeasesUponShardCompletion Clean up leases for shards that we've finished processing (don't wait
      *        until they expire)
+     * @param ignoreUnexpectedChildShards Ignore child shards with open parents
      * @param shardSyncIdleTimeMillis Time between tasks to sync leases and Kinesis shards
      * @param metricsFactory Metrics factory
      * @param executorService ExecutorService to execute the shard sync tasks
@@ -63,6 +65,7 @@ class ShardSyncTaskManager {
             final ILeaseManager<KinesisClientLease> leaseManager,
             final InitialPositionInStreamExtended initialPositionInStream,
             final boolean cleanupLeasesUponShardCompletion,
+            final boolean ignoreUnexpectedChildShards,
             final long shardSyncIdleTimeMillis,
             final IMetricsFactory metricsFactory,
             ExecutorService executorService) {
@@ -70,6 +73,7 @@ class ShardSyncTaskManager {
         this.leaseManager = leaseManager;
         this.metricsFactory = metricsFactory;
         this.cleanupLeasesUponShardCompletion = cleanupLeasesUponShardCompletion;
+        this.ignoreUnexpectedChildShards = ignoreUnexpectedChildShards;
         this.shardSyncIdleTimeMillis = shardSyncIdleTimeMillis;
         this.executorService = executorService;
         this.initialPositionInStream = initialPositionInStream;
@@ -99,6 +103,7 @@ class ShardSyncTaskManager {
                             leaseManager,
                             initialPositionInStream,
                             cleanupLeasesUponShardCompletion,
+                            ignoreUnexpectedChildShards,
                             shardSyncIdleTimeMillis), metricsFactory);
             future = executorService.submit(currentTask);
             submittedNewTask = true;
diff --git a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncer.java b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncer.java
index 52944200..aa8a5841 100644
--- a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncer.java
+++ b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncer.java
@@ -62,42 +62,47 @@ class ShardSyncer {
             InitialPositionInStreamExtended initialPositionInStream,
             boolean cleanupLeasesOfCompletedShards)
         throws DependencyException, InvalidStateException, ProvisionedThroughputException, KinesisClientLibIOException {
-        syncShardLeases(kinesisProxy, leaseManager, initialPositionInStream, cleanupLeasesOfCompletedShards);
+        syncShardLeases(kinesisProxy, leaseManager, initialPositionInStream, cleanupLeasesOfCompletedShards, false);
     }
 
     /**
      * Check and create leases for any new shards (e.g. following a reshard operation).
-     * 
+     *
      * @param kinesisProxy
      * @param leaseManager
      * @param initialPositionInStream
-     * @param expectedClosedShardId If this is not null, we will assert that the shard list we get from Kinesis
-     *        shows this shard to be closed (e.g. parent shard must be closed after a reshard operation).
-     *        If it is open, we assume this is an race condition around a reshard event and throw
-     *        a KinesisClientLibIOException so client can backoff and retry later.
+     * @param cleanupLeasesOfCompletedShards
+     * @param ignoreUnexpectedChildShards
      * @throws DependencyException
      * @throws InvalidStateException
      * @throws ProvisionedThroughputException
      * @throws KinesisClientLibIOException
      */
+    static synchronized void checkAndCreateLeasesForNewShards(IKinesisProxy kinesisProxy,
+            ILeaseManager<KinesisClientLease> leaseManager,
+            InitialPositionInStreamExtended initialPositionInStream,
+            boolean cleanupLeasesOfCompletedShards,
+            boolean ignoreUnexpectedChildShards)
+        throws DependencyException, InvalidStateException, ProvisionedThroughputException, KinesisClientLibIOException {
+        syncShardLeases(kinesisProxy, leaseManager, initialPositionInStream, cleanupLeasesOfCompletedShards, ignoreUnexpectedChildShards);
+    }
+
     static synchronized void checkAndCreateLeasesForNewShards(IKinesisProxy kinesisProxy,
             ILeaseManager<KinesisClientLease> leaseManager,
             InitialPositionInStreamExtended initialPositionInStream,
             boolean cleanupLeasesOfCompletedShards)
         throws DependencyException, InvalidStateException, ProvisionedThroughputException, KinesisClientLibIOException {
-        syncShardLeases(kinesisProxy, leaseManager, initialPositionInStream, cleanupLeasesOfCompletedShards);
+        checkAndCreateLeasesForNewShards(kinesisProxy, leaseManager, initialPositionInStream, cleanupLeasesOfCompletedShards, false);
     }
 
     /**
      * Sync leases with Kinesis shards (e.g. at startup, or when we reach end of a shard).
-     * 
+     *
      * @param kinesisProxy
      * @param leaseManager
-     * @param expectedClosedShardId If this is not null, we will assert that the shard list we get from Kinesis
-     *        does not show this shard to be open (e.g. parent shard must be closed after a reshard operation).
-     *        If it is still open, we assume this is a race condition around a reshard event and
-     *        throw a KinesisClientLibIOException so client can backoff and retry later. If the shard doesn't exist in
-     *        Kinesis at all, we assume this is an old/expired shard and continue with the sync operation.
+     * @param initialPosition
+     * @param cleanupLeasesOfCompletedShards
+     * @param ignoreUnexpectedChildShards
      * @throws DependencyException
      * @throws InvalidStateException
      * @throws ProvisionedThroughputException
@@ -107,18 +112,23 @@ class ShardSyncer {
     private static synchronized void syncShardLeases(IKinesisProxy kinesisProxy,
             ILeaseManager<KinesisClientLease> leaseManager,
             InitialPositionInStreamExtended initialPosition,
-            boolean cleanupLeasesOfCompletedShards)
+            boolean cleanupLeasesOfCompletedShards,
+            boolean ignoreUnexpectedChildShards)
         throws DependencyException, InvalidStateException, ProvisionedThroughputException, KinesisClientLibIOException {
         List<Shard> shards = getShardList(kinesisProxy);
         LOG.debug("Num shards: " + shards.size());
 
         Map<String, Shard> shardIdToShardMap = constructShardIdToShardMap(shards);
         Map<String, Set<String>> shardIdToChildShardIdsMap = constructShardIdToChildShardIdsMap(shardIdToShardMap);
-        assertAllParentShardsAreClosed(shardIdToChildShardIdsMap, shardIdToShardMap);
-        
+        Set<String> inconsistentShardIds = findInconsistentShardIds(shardIdToChildShardIdsMap, shardIdToShardMap);
+        if (!ignoreUnexpectedChildShards) {
+            assertAllParentShardsAreClosed(inconsistentShardIds);
+        }
+
         List<KinesisClientLease> currentLeases = leaseManager.listLeases();
-        
-        List<KinesisClientLease> newLeasesToCreate = determineNewLeasesToCreate(shards, currentLeases, initialPosition);
+
+        List<KinesisClientLease> newLeasesToCreate = determineNewLeasesToCreate(shards, currentLeases, initialPosition,
+                                                                                inconsistentShardIds);
         LOG.debug("Num new leases to create: " + newLeasesToCreate.size());
         for (KinesisClientLease lease : newLeasesToCreate) {
             long startTimeMillis = System.currentTimeMillis();
@@ -130,10 +140,10 @@ class ShardSyncer {
                 MetricsHelper.addSuccessAndLatency("CreateLease", startTimeMillis, success, MetricsLevel.DETAILED);
             }
         }
-        
+
         List<KinesisClientLease> trackedLeases = new ArrayList<>();
         if (currentLeases != null) {
-            trackedLeases.addAll(currentLeases);            
+            trackedLeases.addAll(currentLeases);
         }
         trackedLeases.addAll(newLeasesToCreate);
         cleanupGarbageLeases(shards, trackedLeases, kinesisProxy, leaseManager);
@@ -149,19 +159,39 @@ class ShardSyncer {
 
     /** Helper method to detect a race condition between fetching the shards via paginated DescribeStream calls
      * and a reshard operation.
-     * @param shardIdToChildShardIdsMap
-     * @param shardIdToShardMap
+     * @param inconsistentShardIds
      * @throws KinesisClientLibIOException
      */
-    private static void assertAllParentShardsAreClosed(Map<String, Set<String>> shardIdToChildShardIdsMap,
-            Map<String, Shard> shardIdToShardMap) throws KinesisClientLibIOException {
+    private static void assertAllParentShardsAreClosed(Set<String> inconsistentShardIds)
+        throws KinesisClientLibIOException {
+        if (!inconsistentShardIds.isEmpty()) {
+            String ids = "";
+            for (String id : inconsistentShardIds) {
+                ids += " " + id;
+            }
+            throw new KinesisClientLibIOException(String.valueOf(inconsistentShardIds.size()) + " open child shards (" + ids + ") are inconsistent."
+                                                  + "This can happen due to a race condition between describeStream and a reshard operation.");
+        }
+    }
+
+    /**
+     * Helper method to construct the list of inconsistent shards, which are open shards with non-closed ancestor
+     * parent(s).
+     * @param shardIdToChildShardIdsMap
+     * @param shardIdToShardMap
+     * @return Set of inconsistent open shard ids for shards having open parents.
+     */
+    private static Set<String> findInconsistentShardIds(Map<String, Set<String>> shardIdToChildShardIdsMap,
+            Map<String, Shard> shardIdToShardMap) {
+        Set<String> result = new HashSet<String>();
         for (String parentShardId : shardIdToChildShardIdsMap.keySet()) {
             Shard parentShard = shardIdToShardMap.get(parentShardId);
             if ((parentShardId == null) || (parentShard.getSequenceNumberRange().getEndingSequenceNumber() == null)) {
-                throw new KinesisClientLibIOException("Parent shardId " + parentShardId + " is not closed. " 
-                        + "This can happen due to a race condition between describeStream and a reshard operation.");
+                Set<String> childShardIdsMap = shardIdToChildShardIdsMap.get(parentShardId);
+                result.addAll(childShardIdsMap);
             }
         }
+        return result;
     }
 
     /**
@@ -179,7 +209,7 @@ class ShardSyncer {
     }
 
     /**
-     * Note: this has package level access for testing purposes. 
+     * Note: this has package level access for testing purposes.
      * Useful for asserting that we don't have an incomplete shard list following a reshard operation.
      * We verify that if the shard is present in the shard list, it is closed and its hash key range
      *     is covered by its child shards.
@@ -190,17 +220,17 @@ class ShardSyncer {
      */
     static synchronized void assertClosedShardsAreCoveredOrAbsent(Map<String, Shard> shardIdToShardMap,
             Map<String, Set<String>> shardIdToChildShardIdsMap,
-            Set<String> shardIdsOfClosedShards) throws KinesisClientLibIOException {        
+            Set<String> shardIdsOfClosedShards) throws KinesisClientLibIOException {
         String exceptionMessageSuffix = "This can happen if we constructed the list of shards "
                         + " while a reshard operation was in progress.";
-        
+
         for (String shardId : shardIdsOfClosedShards) {
             Shard shard = shardIdToShardMap.get(shardId);
             if (shard == null) {
                 LOG.info("Shard " + shardId + " is not present in Kinesis anymore.");
                 continue;
             }
-            
+
             String endingSequenceNumber = shard.getSequenceNumberRange().getEndingSequenceNumber();
             if (endingSequenceNumber == null) {
                 throw new KinesisClientLibIOException("Shard " + shardIdsOfClosedShards
@@ -220,7 +250,7 @@ class ShardSyncer {
     private static synchronized void assertHashRangeOfClosedShardIsCovered(Shard closedShard,
             Map<String, Shard> shardIdToShardMap,
             Set<String> childShardIds) throws KinesisClientLibIOException {
-        
+
         BigInteger startingHashKeyOfClosedShard = new BigInteger(closedShard.getHashKeyRange().getStartingHashKey());
         BigInteger endingHashKeyOfClosedShard = new BigInteger(closedShard.getHashKeyRange().getEndingHashKey());
         BigInteger minStartingHashKeyOfChildren = null;
@@ -239,16 +269,16 @@ class ShardSyncer {
                 maxEndingHashKeyOfChildren = endingHashKey;
             }
         }
-        
+
         if ((minStartingHashKeyOfChildren == null) || (maxEndingHashKeyOfChildren == null)
                 || (minStartingHashKeyOfChildren.compareTo(startingHashKeyOfClosedShard) > 0)
                 || (maxEndingHashKeyOfChildren.compareTo(endingHashKeyOfClosedShard) < 0)) {
             throw new KinesisClientLibIOException("Incomplete shard list: hash key range of shard "
                     + closedShard.getShardId() + " is not covered by its child shards.");
         }
-        
+
     }
-    
+
     /**
      * Helper method to construct shardId->setOfChildShardIds map.
      * Note: This has package access for testing purposes only.
@@ -270,7 +300,7 @@ class ShardSyncer {
                 }
                 childShardIds.add(shardId);
             }
-            
+
             String adjacentParentShardId = shard.getAdjacentParentShardId();
             if ((adjacentParentShardId != null) && (shardIdToShardMap.containsKey(adjacentParentShardId))) {
                 Set<String> childShardIds = shardIdToChildShardIdsMap.get(adjacentParentShardId);
@@ -296,8 +326,8 @@ class ShardSyncer {
     /**
      * Determine new leases to create and their initial checkpoint.
      * Note: Package level access only for testing purposes.
-     * 
-     * For each open (no ending sequence number) shard that doesn't already have a lease,
+     *
+     * For each open (no ending sequence number) shard without open parents that doesn't already have a lease,
      * determine if it is a descendent of any shard which is or will be processed (e.g. for which a lease exists):
      * If so, set checkpoint of the shard to TrimHorizon and also create leases for ancestors if needed.
      * If not, set checkpoint of the shard to the initial position specified by the client.
@@ -306,36 +336,43 @@ class ShardSyncer {
      *         we begin processing data from any of its descendants.
      *   * A shard does not start processing data until data from all its parents has been processed.
      * Note, if the initial position is LATEST and a shard has two parents and only one is a descendant - we'll create
-     * leases corresponding to both the parents - the parent shard which is not a descendant will have  
+     * leases corresponding to both the parents - the parent shard which is not a descendant will have
      * its checkpoint set to Latest.
-     * 
+     *
      * We assume that if there is an existing lease for a shard, then either:
      *   * we have previously created a lease for its parent (if it was needed), or
      *   * the parent shard has expired.
-     * 
+     *
      * For example:
      * Shard structure (each level depicts a stream segment):
-     * 0 1 2 3 4 5- shards till epoch 102
-     * \ / \ / | |
-     * 6 7 4 5- shards from epoch 103 - 205
-     * \ / | /\
-     * 8 4 9 10 - shards from epoch 206 (open - no ending sequenceNumber)
+     * 0 1 2 3 4   5   - shards till epoch 102
+     * \ / \ / |   |
+     *  6   7  4   5   - shards from epoch 103 - 205
+     *   \ /   |  / \
+     *    8    4 9  10 - shards from epoch 206 (open - no ending sequenceNumber)
      * Current leases: (3, 4, 5)
      * New leases to create: (2, 6, 7, 8, 9, 10)
-     * 
+     *
      * The leases returned are sorted by the starting sequence number - following the same order
      * when persisting the leases in DynamoDB will ensure that we recover gracefully if we fail
      * before creating all the leases.
-     * 
+     *
+     * If a shard has no existing lease, is open, and is a descendant of a parent which is still open, we ignore it
+     * here; this happens when the list of shards is inconsistent, which could be due to pagination delay for very
+     * high shard count streams (i.e., dynamodb streams for tables with thousands of partitions).  This can only
+     * currently happen here if ignoreUnexpectedChildShards was true in syncShardleases.
+     *
      * @param shards List of all shards in Kinesis (we'll create new leases based on this set)
      * @param currentLeases List of current leases
      * @param initialPosition One of LATEST, TRIM_HORIZON, or AT_TIMESTAMP. We'll start fetching records from that
      *        location in the shard (when an application starts up for the first time - and there are no checkpoints).
+     * @param inconsistentShardIds Set of child shard ids having open parents.
      * @return List of new leases to create sorted by starting sequenceNumber of the corresponding shard
      */
     static List<KinesisClientLease> determineNewLeasesToCreate(List<Shard> shards,
             List<KinesisClientLease> currentLeases,
-            InitialPositionInStreamExtended initialPosition) {
+            InitialPositionInStreamExtended initialPosition,
+            Set<String> inconsistentShardIds) {
         Map<String, KinesisClientLease> shardIdToNewLeaseMap = new HashMap<String, KinesisClientLease>();
         Map<String, Shard> shardIdToShardMapOfAllKinesisShards = constructShardIdToShardMap(shards);
 
@@ -354,6 +391,8 @@ class ShardSyncer {
             LOG.debug("Evaluating leases for open shard " + shardId + " and its ancestors.");
             if (shardIdsOfCurrentLeases.contains(shardId)) {
                 LOG.debug("Lease for shardId " + shardId + " already exists. Not creating a lease");
+            } else if (inconsistentShardIds.contains(shardId)) {
+                LOG.info("shardId " + shardId + " is an inconsistent child.  Not creating a lease");
             } else {
                 LOG.debug("Need to create a lease for shardId " + shardId);
                 KinesisClientLease newLease = newKCLLease(shard);
@@ -407,12 +446,23 @@ class ShardSyncer {
         return newLeasesToCreate;
     }
 
+    /**
+     * Determine new leases to create and their initial checkpoint.
+     * Note: Package level access only for testing purposes.
+     */
+    static List<KinesisClientLease> determineNewLeasesToCreate(List<Shard> shards,
+            List<KinesisClientLease> currentLeases,
+            InitialPositionInStreamExtended initialPosition) {
+        Set<String> inconsistentShardIds = new HashSet<String>();
+        return determineNewLeasesToCreate(shards, currentLeases, initialPosition, inconsistentShardIds);
+    }
+
     /**
      * Note: Package level access for testing purposes only.
      * Check if this shard is a descendant of a shard that is (or will be) processed.
      * Create leases for the ancestors of this shard as required.
      * See javadoc of determineNewLeasesToCreate() for rules and example.
-     * 
+     *
      * @param shardId The shardId to check.
      * @param initialPosition One of LATEST, TRIM_HORIZON, or AT_TIMESTAMP. We'll start fetching records from that
      *        location in the shard (when an application starts up for the first time - and there are no checkpoints).
@@ -429,7 +479,7 @@ class ShardSyncer {
             Map<String, Shard> shardIdToShardMapOfAllKinesisShards,
             Map<String, KinesisClientLease> shardIdToLeaseMapOfNewShards,
             Map<String, Boolean> memoizationContext) {
-        
+
         Boolean previousValue = memoizationContext.get(shardId);
         if (previousValue != null) {
             return previousValue;
@@ -509,7 +559,7 @@ class ShardSyncer {
      * Helper method to get parent shardIds of the current shard - includes the parent shardIds if:
      * a/ they are not null
      * b/ if they exist in the current shard map (i.e. haven't expired)
-     * 
+     *
      * @param shard Will return parents of this shard
      * @param shardIdToShardMapOfAllKinesisShards ShardId->Shard map containing all shards obtained via DescribeStream.
      * @return Set of parentShardIds
@@ -528,18 +578,18 @@ class ShardSyncer {
     }
 
     /**
-     * Delete leases corresponding to shards that no longer exist in the stream. 
+     * Delete leases corresponding to shards that no longer exist in the stream.
      * Current scheme: Delete a lease if:
      *   * the corresponding shard is not present in the list of Kinesis shards, AND
      *   * the parentShardIds listed in the lease are also not present in the list of Kinesis shards.
      * @param shards List of all Kinesis shards (assumed to be a consistent snapshot - when stream is in Active state).
-     * @param trackedLeases List of 
+     * @param trackedLeases List of
      * @param kinesisProxy Kinesis proxy (used to get shard list)
-     * @param leaseManager 
+     * @param leaseManager
      * @throws KinesisClientLibIOException Thrown if we couldn't get a fresh shard list from Kinesis.
-     * @throws ProvisionedThroughputException 
-     * @throws InvalidStateException 
-     * @throws DependencyException 
+     * @throws ProvisionedThroughputException
+     * @throws InvalidStateException
+     * @throws DependencyException
      */
     private static void cleanupGarbageLeases(List<Shard> shards,
             List<KinesisClientLease> trackedLeases,
@@ -550,7 +600,7 @@ class ShardSyncer {
         for (Shard shard : shards) {
             kinesisShards.add(shard.getShardId());
         }
-        
+
         // Check if there are leases for non-existent shards
         List<KinesisClientLease> garbageLeases = new ArrayList<>();
         for (KinesisClientLease lease : trackedLeases) {
@@ -558,10 +608,10 @@ class ShardSyncer {
                 garbageLeases.add(lease);
             }
         }
-        
+
         if (!garbageLeases.isEmpty()) {
             LOG.info("Found " + garbageLeases.size()
-                    + " candidate leases for cleanup. Refreshing list of" 
+                    + " candidate leases for cleanup. Refreshing list of"
                     + " Kinesis shards to pick up recent/latest shards");
             List<Shard> currentShardList = getShardList(kinesisProxy);
             Set<String> currentKinesisShardIds = new HashSet<>();
@@ -577,12 +627,12 @@ class ShardSyncer {
                 }
             }
         }
-        
+
     }
 
     /**
      * Note: This method has package level access, solely for testing purposes.
-     * 
+     *
      * @param lease Candidate shard we are considering for deletion.
      * @param currentKinesisShardIds
      * @return true if neither the shard (corresponding to the lease), nor its parents are present in
@@ -593,16 +643,16 @@ class ShardSyncer {
     static boolean isCandidateForCleanup(KinesisClientLease lease, Set<String> currentKinesisShardIds)
         throws KinesisClientLibIOException {
         boolean isCandidateForCleanup = true;
-        
+
         if (currentKinesisShardIds.contains(lease.getLeaseKey())) {
             isCandidateForCleanup = false;
         } else {
             LOG.info("Found lease for non-existent shard: " + lease.getLeaseKey() + ". Checking its parent shards");
             Set<String> parentShardIds = lease.getParentShardIds();
             for (String parentShardId : parentShardIds) {
-                
+
                 // Throw an exception if the parent shard exists (but the child does not).
-                // This may be a (rare) race condition between fetching the shard list and Kinesis expiring shards. 
+                // This may be a (rare) race condition between fetching the shard list and Kinesis expiring shards.
                 if (currentKinesisShardIds.contains(parentShardId)) {
                     String message =
                             "Parent shard " + parentShardId + " exists but not the child shard "
@@ -615,14 +665,14 @@ class ShardSyncer {
 
         return isCandidateForCleanup;
     }
-    
+
     /**
      * Private helper method.
      * Clean up leases for shards that meet the following criteria:
      * a/ the shard has been fully processed (checkpoint is set to SHARD_END)
      * b/ we've begun processing all the child shards: we have leases for all child shards and their checkpoint is not
      *      TRIM_HORIZON.
-     * 
+     *
      * @param currentLeases List of leases we evaluate for clean up
      * @param shardIdToShardMap Map of shardId->Shard (assumed to include all Kinesis shards)
      * @param shardIdToChildShardIdsMap Map of shardId->childShardIds (assumed to include all Kinesis shards)
@@ -664,22 +714,22 @@ class ShardSyncer {
                     cleanupLeaseForClosedShard(closedShardId, childShardIds, trackedLeaseMap, leaseManager);
                 }
             }
-        }        
+        }
     }
 
-    /** 
+    /**
      * Delete lease for the closed shard. Rules for deletion are:
      * a/ the checkpoint for the closed shard is SHARD_END,
      * b/ there are leases for all the childShardIds and their checkpoint is NOT TRIM_HORIZON
      * Note: This method has package level access solely for testing purposes.
-     * 
+     *
      * @param closedShardId Identifies the closed shard
      * @param childShardIds ShardIds of children of the closed shard
      * @param trackedLeases shardId->KinesisClientLease map with all leases we are tracking (should not be null)
-     * @param leaseManager 
-     * @throws ProvisionedThroughputException 
-     * @throws InvalidStateException 
-     * @throws DependencyException 
+     * @param leaseManager
+     * @throws ProvisionedThroughputException
+     * @throws InvalidStateException
+     * @throws DependencyException
      */
     static synchronized void cleanupLeaseForClosedShard(String closedShardId,
             Set<String> childShardIds,
@@ -688,14 +738,14 @@ class ShardSyncer {
         throws DependencyException, InvalidStateException, ProvisionedThroughputException {
         KinesisClientLease leaseForClosedShard = trackedLeases.get(closedShardId);
         List<KinesisClientLease> childShardLeases = new ArrayList<>();
-        
+
         for (String childShardId : childShardIds) {
             KinesisClientLease childLease = trackedLeases.get(childShardId);
             if (childLease != null) {
                 childShardLeases.add(childLease);
             }
         }
-        
+
         if ((leaseForClosedShard != null)
                 && (leaseForClosedShard.getCheckpoint().equals(ExtendedSequenceNumber.SHARD_END))
                 && (childShardLeases.size() == childShardIds.size())) {
@@ -706,7 +756,7 @@ class ShardSyncer {
                     break;
                 }
             }
-            
+
             if (okayToDelete) {
                 LOG.info("Deleting lease for shard " + leaseForClosedShard.getLeaseKey()
                         + " as it has been completely processed and processing of child shards has begun.");
@@ -718,7 +768,7 @@ class ShardSyncer {
     /**
      * Helper method to create a new KinesisClientLease POJO for a shard.
      * Note: Package level access only for testing purposes
-     * 
+     *
      * @param shard
      * @return
      */
@@ -740,7 +790,7 @@ class ShardSyncer {
 
     /**
      * Helper method to construct a shardId->Shard map for the specified list of shards.
-     * 
+     *
      * @param shards List of shards
      * @return ShardId->Shard map
      */
@@ -755,7 +805,7 @@ class ShardSyncer {
     /**
      * Helper method to return all the open shards for a stream.
      * Note: Package level access only for testing purposes.
-     * 
+     *
      * @param allShards All shards returved via DescribeStream. We assume this to represent a consistent shard list.
      * @return List of open shards (shards at the tip of the stream) - may include shards that are not yet active.
      */
@@ -773,7 +823,7 @@ class ShardSyncer {
 
     private static ExtendedSequenceNumber convertToCheckpoint(InitialPositionInStreamExtended position) {
         ExtendedSequenceNumber checkpoint = null;
-        
+
         if (position.getInitialPositionInStream().equals(InitialPositionInStream.TRIM_HORIZON)) {
             checkpoint = ExtendedSequenceNumber.TRIM_HORIZON;
         } else if (position.getInitialPositionInStream().equals(InitialPositionInStream.LATEST)) {
@@ -781,10 +831,10 @@ class ShardSyncer {
         } else if (position.getInitialPositionInStream().equals(InitialPositionInStream.AT_TIMESTAMP)) {
             checkpoint = ExtendedSequenceNumber.AT_TIMESTAMP;
         }
-        
+
         return checkpoint;
     }
-    
+
     /** Helper class to compare leases based on starting sequence number of the corresponding shards.
      *
      */
@@ -794,7 +844,7 @@ class ShardSyncer {
         private static final long serialVersionUID = 1L;
 
         private final Map<String, Shard> shardIdToShardMap;
-        
+
         /**
          * @param shardIdToShardMapOfAllKinesisShards
          */
@@ -808,7 +858,7 @@ class ShardSyncer {
          * We assume that lease1 and lease2 are:
          *     a/ not null,
          *     b/ shards (if found) have non-null starting sequence numbers
-         * 
+         *
          * {@inheritDoc}
          */
         @Override
@@ -818,23 +868,23 @@ class ShardSyncer {
             String shardId2 = lease2.getLeaseKey();
             Shard shard1 = shardIdToShardMap.get(shardId1);
             Shard shard2 = shardIdToShardMap.get(shardId2);
-            
+
             // If we found shards for the two leases, use comparison of the starting sequence numbers
             if ((shard1 != null) && (shard2 != null)) {
                 BigInteger sequenceNumber1 =
                         new BigInteger(shard1.getSequenceNumberRange().getStartingSequenceNumber());
                 BigInteger sequenceNumber2 =
                         new BigInteger(shard2.getSequenceNumberRange().getStartingSequenceNumber());
-                result = sequenceNumber1.compareTo(sequenceNumber2);                
+                result = sequenceNumber1.compareTo(sequenceNumber2);
             }
-            
+
             if (result == 0) {
                 result = shardId1.compareTo(shardId2);
             }
-            
+
             return result;
         }
-                
+
     }
 
 }
diff --git a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShutdownTask.java b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShutdownTask.java
index f56033a8..1d49817e 100644
--- a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShutdownTask.java
+++ b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShutdownTask.java
@@ -44,6 +44,7 @@ class ShutdownTask implements ITask {
     private final ILeaseManager<KinesisClientLease> leaseManager;
     private final InitialPositionInStreamExtended initialPositionInStream;
     private final boolean cleanupLeasesOfCompletedShards;
+    private final boolean ignoreUnexpectedChildShards;
     private final TaskType taskType = TaskType.SHUTDOWN;
     private final long backoffTimeMillis;
     private final GetRecordsRetrievalStrategy getRecordsRetrievalStrategy;
@@ -59,8 +60,9 @@ class ShutdownTask implements ITask {
                  IKinesisProxy kinesisProxy,
                  InitialPositionInStreamExtended initialPositionInStream,
                  boolean cleanupLeasesOfCompletedShards,
+                 boolean ignoreUnexpectedChildShards,
                  ILeaseManager<KinesisClientLease> leaseManager,
-                 long backoffTimeMillis, 
+                 long backoffTimeMillis,
                  GetRecordsRetrievalStrategy getRecordsRetrievalStrategy) {
         this.shardInfo = shardInfo;
         this.recordProcessor = recordProcessor;
@@ -69,6 +71,7 @@ class ShutdownTask implements ITask {
         this.kinesisProxy = kinesisProxy;
         this.initialPositionInStream = initialPositionInStream;
         this.cleanupLeasesOfCompletedShards = cleanupLeasesOfCompletedShards;
+        this.ignoreUnexpectedChildShards = ignoreUnexpectedChildShards;
         this.leaseManager = leaseManager;
         this.backoffTimeMillis = backoffTimeMillis;
         this.getRecordsRetrievalStrategy = getRecordsRetrievalStrategy;
@@ -77,7 +80,7 @@ class ShutdownTask implements ITask {
     /*
      * Invokes RecordProcessor shutdown() API.
      * (non-Javadoc)
-     * 
+     *
      * @see com.amazonaws.services.kinesis.clientlibrary.lib.worker.ITask#call()
      */
     @Override
@@ -127,7 +130,8 @@ class ShutdownTask implements ITask {
                 ShardSyncer.checkAndCreateLeasesForNewShards(kinesisProxy,
                         leaseManager,
                         initialPositionInStream,
-                        cleanupLeasesOfCompletedShards);
+                        cleanupLeasesOfCompletedShards,
+                        ignoreUnexpectedChildShards);
                 LOG.debug("Finished checking for child shards of shard " + shardInfo.getShardId());
             }
 
@@ -152,7 +156,7 @@ class ShutdownTask implements ITask {
 
     /*
      * (non-Javadoc)
-     * 
+     *
      * @see com.amazonaws.services.kinesis.clientlibrary.lib.worker.ITask#getTaskType()
      */
     @Override
diff --git a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/Worker.java b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/Worker.java
index 3cfb9f2f..0906bbb0 100644
--- a/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/Worker.java
+++ b/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/Worker.java
@@ -103,6 +103,7 @@ public class Worker implements Runnable {
     // info, value is ShardConsumer.
     private ConcurrentMap<ShardInfo, ShardConsumer> shardInfoShardConsumerMap = new ConcurrentHashMap<ShardInfo, ShardConsumer>();
     private final boolean cleanupLeasesUponShardCompletion;
+    private final boolean ignoreUnexpectedChildShards;
 
     private final boolean skipShardSyncAtWorkerInitializationIfLeasesExist;
 
@@ -253,7 +254,8 @@ public class Worker implements Runnable {
                         config.shouldValidateSequenceNumberBeforeCheckpointing(),
                         config.getInitialPositionInStreamExtended()),
                 config.getInitialPositionInStreamExtended(), config.getParentShardPollIntervalMillis(),
-                config.getShardSyncIntervalMillis(), config.shouldCleanupLeasesUponShardCompletion(), null,
+                config.getShardSyncIntervalMillis(), config.shouldCleanupLeasesUponShardCompletion(),
+                config.shouldIgnoreUnexpectedChildShards(), null,
                 new KinesisClientLibLeaseCoordinator(
                         new KinesisClientLeaseManager(config.getTableName(), dynamoDBClient),
                         config.getWorkerIdentifier(),
@@ -318,6 +320,8 @@ public class Worker implements Runnable {
      *            Time between tasks to sync leases and Kinesis shards
      * @param cleanupLeasesUponShardCompletion
      *            Clean up shards we've finished processing (don't wait till they expire in Kinesis)
+     * @param ignoreUnexpectedChildShards
+     *            Ignore child shards with open parents
      * @param checkpoint
      *            Used to get/set checkpoints
      * @param leaseCoordinator
@@ -335,14 +339,14 @@ public class Worker implements Runnable {
     // CHECKSTYLE:IGNORE ParameterNumber FOR NEXT 10 LINES
     Worker(String applicationName, IRecordProcessorFactory recordProcessorFactory, StreamConfig streamConfig,
             InitialPositionInStreamExtended initialPositionInStream, long parentShardPollIntervalMillis,
-            long shardSyncIdleTimeMillis, boolean cleanupLeasesUponShardCompletion, ICheckpoint checkpoint,
-            KinesisClientLibLeaseCoordinator leaseCoordinator, ExecutorService execService,
+            long shardSyncIdleTimeMillis, boolean cleanupLeasesUponShardCompletion, boolean ignoreUnexpectedChildShards,
+            ICheckpoint checkpoint, KinesisClientLibLeaseCoordinator leaseCoordinator, ExecutorService execService,
             IMetricsFactory metricsFactory, long taskBackoffTimeMillis, long failoverTimeMillis,
             boolean skipShardSyncAtWorkerInitializationIfLeasesExist, ShardPrioritization shardPrioritization) {
         this(applicationName, recordProcessorFactory, streamConfig, initialPositionInStream, parentShardPollIntervalMillis,
-                shardSyncIdleTimeMillis, cleanupLeasesUponShardCompletion, checkpoint, leaseCoordinator, execService,
-                metricsFactory, taskBackoffTimeMillis, failoverTimeMillis, skipShardSyncAtWorkerInitializationIfLeasesExist,
-                shardPrioritization, Optional.empty(), Optional.empty());
+                shardSyncIdleTimeMillis, cleanupLeasesUponShardCompletion, ignoreUnexpectedChildShards, checkpoint,
+                leaseCoordinator, execService, metricsFactory, taskBackoffTimeMillis, failoverTimeMillis,
+                skipShardSyncAtWorkerInitializationIfLeasesExist, shardPrioritization, Optional.empty(), Optional.empty());
     }
 
 
@@ -363,6 +367,8 @@ public class Worker implements Runnable {
      *            Time between tasks to sync leases and Kinesis shards
      * @param cleanupLeasesUponShardCompletion
      *            Clean up shards we've finished processing (don't wait till they expire in Kinesis)
+     * @param ignoreUnexpectedChildShards
+     *            Ignore child shards with open parents
      * @param checkpoint
      *            Used to get/set checkpoints
      * @param leaseCoordinator
@@ -384,8 +390,8 @@ public class Worker implements Runnable {
     // CHECKSTYLE:IGNORE ParameterNumber FOR NEXT 10 LINES
     Worker(String applicationName, IRecordProcessorFactory recordProcessorFactory, StreamConfig streamConfig,
            InitialPositionInStreamExtended initialPositionInStream, long parentShardPollIntervalMillis,
-           long shardSyncIdleTimeMillis, boolean cleanupLeasesUponShardCompletion, ICheckpoint checkpoint,
-           KinesisClientLibLeaseCoordinator leaseCoordinator, ExecutorService execService,
+           long shardSyncIdleTimeMillis, boolean cleanupLeasesUponShardCompletion, boolean ignoreUnexpectedChildShards,
+           ICheckpoint checkpoint, KinesisClientLibLeaseCoordinator leaseCoordinator, ExecutorService execService,
            IMetricsFactory metricsFactory, long taskBackoffTimeMillis, long failoverTimeMillis,
            boolean skipShardSyncAtWorkerInitializationIfLeasesExist, ShardPrioritization shardPrioritization,
            Optional<Integer> retryGetRecordsInSeconds, Optional<Integer> maxGetRecordsThreadPool) {
@@ -395,14 +401,15 @@ public class Worker implements Runnable {
         this.initialPosition = initialPositionInStream;
         this.parentShardPollIntervalMillis = parentShardPollIntervalMillis;
         this.cleanupLeasesUponShardCompletion = cleanupLeasesUponShardCompletion;
+        this.ignoreUnexpectedChildShards = ignoreUnexpectedChildShards;
         this.checkpointTracker = checkpoint != null ? checkpoint : leaseCoordinator;
         this.idleTimeInMilliseconds = streamConfig.getIdleTimeInMilliseconds();
         this.executorService = execService;
         this.leaseCoordinator = leaseCoordinator;
         this.metricsFactory = metricsFactory;
         this.controlServer = new ShardSyncTaskManager(streamConfig.getStreamProxy(), leaseCoordinator.getLeaseManager(),
-                initialPositionInStream, cleanupLeasesUponShardCompletion, shardSyncIdleTimeMillis, metricsFactory,
-                executorService);
+                initialPositionInStream, cleanupLeasesUponShardCompletion, ignoreUnexpectedChildShards, shardSyncIdleTimeMillis,
+                metricsFactory, executorService);
         this.taskBackoffTimeMillis = taskBackoffTimeMillis;
         this.failoverTimeMillis = failoverTimeMillis;
         this.skipShardSyncAtWorkerInitializationIfLeasesExist = skipShardSyncAtWorkerInitializationIfLeasesExist;
@@ -494,7 +501,8 @@ public class Worker implements Runnable {
                         || leaseCoordinator.getLeaseManager().isLeaseTableEmpty()) {
                     LOG.info("Syncing Kinesis shard info");
                     ShardSyncTask shardSyncTask = new ShardSyncTask(streamConfig.getStreamProxy(),
-                            leaseCoordinator.getLeaseManager(), initialPosition, cleanupLeasesUponShardCompletion, 0L);
+                            leaseCoordinator.getLeaseManager(), initialPosition, cleanupLeasesUponShardCompletion,
+                            ignoreUnexpectedChildShards, 0L);
                     result = new MetricsCollectingTaskDecorator(shardSyncTask, metricsFactory).call();
                 } else {
                     LOG.info("Skipping shard sync per config setting (and lease table is not empty)");
@@ -792,7 +800,7 @@ public class Worker implements Runnable {
     /**
      * Returns whether worker can shutdown immediately. Note that this method is called from Worker's {{@link #run()}
      * method before every loop run, so method must do minimum amount of work to not impact shard processing timings.
-     * 
+     *
      * @return Whether worker should shutdown immediately.
      */
     @VisibleForTesting
@@ -844,7 +852,7 @@ public class Worker implements Runnable {
 
         return new ShardConsumer(shardInfo, streamConfig, checkpointTracker, recordProcessor,
                 leaseCoordinator.getLeaseManager(), parentShardPollIntervalMillis, cleanupLeasesUponShardCompletion,
-                executorService, metricsFactory, taskBackoffTimeMillis,
+                ignoreUnexpectedChildShards, executorService, metricsFactory, taskBackoffTimeMillis,
                 skipShardSyncAtWorkerInitializationIfLeasesExist, retryGetRecordsInSeconds, maxGetRecordsThreadPool);
 
     }
@@ -1164,10 +1172,10 @@ public class Worker implements Runnable {
 
         /**
          * Provides logic how to prioritize shard processing.
-         * 
+         *
          * @param shardPrioritization
          *            shardPrioritization is responsible to order shards before processing
-         * 
+         *
          * @return A reference to this updated object so that method calls can be chained together.
          */
         public Builder shardPrioritization(ShardPrioritization shardPrioritization) {
@@ -1255,6 +1263,7 @@ public class Worker implements Runnable {
                     config.getParentShardPollIntervalMillis(),
                     config.getShardSyncIntervalMillis(),
                     config.shouldCleanupLeasesUponShardCompletion(),
+                    config.shouldIgnoreUnexpectedChildShards(),
                     null,
                     new KinesisClientLibLeaseCoordinator(new KinesisClientLeaseManager(config.getTableName(),
                             dynamoDBClient),
diff --git a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfigurationTest.java b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfigurationTest.java
index cfa8be10..30e6ce87 100644
--- a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfigurationTest.java
+++ b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfigurationTest.java
@@ -77,6 +77,7 @@ public class KinesisClientLibConfigurationTest {
                         TEST_VALUE_LONG,
                         TEST_VALUE_LONG,
                         true,
+                        false,
                         new ClientConfiguration(),
                         new ClientConfiguration(),
                         new ClientConfiguration(),
@@ -116,6 +117,7 @@ public class KinesisClientLibConfigurationTest {
                                 longValues[2],
                                 longValues[3],
                                 true,
+                                false,
                                 new ClientConfiguration(),
                                 new ClientConfiguration(),
                                 new ClientConfiguration(),
@@ -151,6 +153,7 @@ public class KinesisClientLibConfigurationTest {
                                 TEST_VALUE_LONG,
                                 TEST_VALUE_LONG,
                                 true,
+                                false,
                                 new ClientConfiguration(),
                                 new ClientConfiguration(),
                                 new ClientConfiguration(),
@@ -315,6 +318,7 @@ public class KinesisClientLibConfigurationTest {
                         TEST_VALUE_LONG,
                         TEST_VALUE_LONG,
                         true,
+                        false,
                         new ClientConfiguration(),
                         new ClientConfiguration(),
                         new ClientConfiguration(),
diff --git a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumerTest.java b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumerTest.java
index a3f786a6..3eef97da 100644
--- a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumerTest.java
+++ b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardConsumerTest.java
@@ -87,6 +87,7 @@ public class ShardConsumerTest {
     private final long taskBackoffTimeMillis = 500L;
     private final long parentShardPollIntervalMillis = 50L;
     private final boolean cleanupLeasesOfCompletedShards = true;
+    private final boolean ignoreUnexpectedChildShards = false;
     // We don't want any of these tests to run checkpoint validation
     private final boolean skipCheckpointValidationValue = false;
     private static final InitialPositionInStreamExtended INITIAL_POSITION_LATEST =
@@ -134,6 +135,7 @@ public class ShardConsumerTest {
                         null,
                         parentShardPollIntervalMillis,
                         cleanupLeasesOfCompletedShards,
+                        ignoreUnexpectedChildShards,
                         executorService,
                         metricsFactory,
                         taskBackoffTimeMillis,
@@ -182,6 +184,7 @@ public class ShardConsumerTest {
                         null,
                         parentShardPollIntervalMillis,
                         cleanupLeasesOfCompletedShards,
+                        ignoreUnexpectedChildShards,
                         spyExecutorService,
                         metricsFactory,
                         taskBackoffTimeMillis,
@@ -223,6 +226,7 @@ public class ShardConsumerTest {
                         null,
                         parentShardPollIntervalMillis,
                         cleanupLeasesOfCompletedShards,
+                        ignoreUnexpectedChildShards,
                         executorService,
                         metricsFactory,
                         taskBackoffTimeMillis,
@@ -318,6 +322,7 @@ public class ShardConsumerTest {
                         leaseManager,
                         parentShardPollIntervalMillis,
                         cleanupLeasesOfCompletedShards,
+                        ignoreUnexpectedChildShards,
                         executorService,
                         metricsFactory,
                         taskBackoffTimeMillis,
@@ -421,6 +426,7 @@ public class ShardConsumerTest {
                         leaseManager,
                         parentShardPollIntervalMillis,
                         cleanupLeasesOfCompletedShards,
+                        ignoreUnexpectedChildShards,
                         executorService,
                         metricsFactory,
                         taskBackoffTimeMillis,
@@ -483,6 +489,7 @@ public class ShardConsumerTest {
                         null,
                         parentShardPollIntervalMillis,
                         cleanupLeasesOfCompletedShards,
+                        ignoreUnexpectedChildShards,
                         executorService,
                         metricsFactory,
                         taskBackoffTimeMillis,
@@ -530,6 +537,7 @@ public class ShardConsumerTest {
                         null,
                         parentShardPollIntervalMillis,
                         cleanupLeasesOfCompletedShards,
+                        ignoreUnexpectedChildShards,
                         executorService,
                         metricsFactory,
                         taskBackoffTimeMillis,
@@ -558,6 +566,7 @@ public class ShardConsumerTest {
                         null,
                         parentShardPollIntervalMillis,
                         cleanupLeasesOfCompletedShards,
+                        ignoreUnexpectedChildShards,
                         executorService,
                         metricsFactory,
                         taskBackoffTimeMillis,
diff --git a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTaskIntegrationTest.java b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTaskIntegrationTest.java
index 307596e3..5fab6375 100644
--- a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTaskIntegrationTest.java
+++ b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShardSyncTaskIntegrationTest.java
@@ -105,7 +105,7 @@ public class ShardSyncTaskIntegrationTest {
 
     /**
      * Test method for call().
-     * 
+     *
      * @throws CapacityExceededException
      * @throws DependencyException
      * @throws InvalidStateException
@@ -123,7 +123,7 @@ public class ShardSyncTaskIntegrationTest {
         ShardSyncTask syncTask = new ShardSyncTask(kinesisProxy,
                 leaseManager,
                 InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.LATEST),
-                false,
+                false, false,
                 0L);
         syncTask.call();
         List<KinesisClientLease> leases = leaseManager.listLeases();
diff --git a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShutdownTaskTest.java b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShutdownTaskTest.java
index 5d91c698..5781826e 100644
--- a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShutdownTaskTest.java
+++ b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/ShutdownTaskTest.java
@@ -100,6 +100,7 @@ public class ShutdownTaskTest {
         IKinesisProxy kinesisProxy = mock(IKinesisProxy.class);
         ILeaseManager<KinesisClientLease> leaseManager = mock(KinesisClientLeaseManager.class);
         boolean cleanupLeasesOfCompletedShards = false;
+        boolean ignoreUnexpectedChildShards = false;
         ShutdownTask task = new ShutdownTask(defaultShardInfo,
                 defaultRecordProcessor,
                 checkpointer,
@@ -107,6 +108,7 @@ public class ShutdownTaskTest {
                 kinesisProxy,
                 INITIAL_POSITION_TRIM_HORIZON,
                 cleanupLeasesOfCompletedShards,
+                ignoreUnexpectedChildShards,
                 leaseManager,
                 TASK_BACKOFF_TIME_MILLIS,
                 getRecordsRetrievalStrategy);
@@ -126,6 +128,7 @@ public class ShutdownTaskTest {
         when(kinesisProxy.getShardList()).thenReturn(null);
         ILeaseManager<KinesisClientLease> leaseManager = mock(KinesisClientLeaseManager.class);
         boolean cleanupLeasesOfCompletedShards = false;
+        boolean ignoreUnexpectedChildShards = false;
         ShutdownTask task = new ShutdownTask(defaultShardInfo,
                 defaultRecordProcessor,
                 checkpointer,
@@ -133,6 +136,7 @@ public class ShutdownTaskTest {
                 kinesisProxy,
                 INITIAL_POSITION_TRIM_HORIZON,
                 cleanupLeasesOfCompletedShards,
+                ignoreUnexpectedChildShards,
                 leaseManager,
                 TASK_BACKOFF_TIME_MILLIS,
                 getRecordsRetrievalStrategy);
@@ -147,7 +151,7 @@ public class ShutdownTaskTest {
      */
     @Test
     public final void testGetTaskType() {
-        ShutdownTask task = new ShutdownTask(null, null, null, null, null, null, false, null, 0, getRecordsRetrievalStrategy);
+        ShutdownTask task = new ShutdownTask(null, null, null, null, null, null, false, false, null, 0, getRecordsRetrievalStrategy);
         Assert.assertEquals(TaskType.SHUTDOWN, task.getTaskType());
     }
 
diff --git a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/WorkerTest.java b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/WorkerTest.java
index 5913bf0d..17444fee 100644
--- a/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/WorkerTest.java
+++ b/src/test/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/WorkerTest.java
@@ -119,6 +119,7 @@ public class WorkerTest {
     private final long parentShardPollIntervalMillis = 5L;
     private final long shardSyncIntervalMillis = 5L;
     private final boolean cleanupLeasesUponShardCompletion = true;
+    private final boolean ignoreUnexpectedChildShards = false;
     // We don't want any of these tests to run checkpoint validation
     private final boolean skipCheckpointValidationValue = false;
     private static final InitialPositionInStreamExtended INITIAL_POSITION_LATEST =
@@ -156,7 +157,7 @@ public class WorkerTest {
     private TaskResult taskResult;
 
     // CHECKSTYLE:IGNORE AnonInnerLengthCheck FOR NEXT 50 LINES
-    private static final com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory SAMPLE_RECORD_PROCESSOR_FACTORY = 
+    private static final com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory SAMPLE_RECORD_PROCESSOR_FACTORY =
             new com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory() {
 
         @Override
@@ -189,8 +190,8 @@ public class WorkerTest {
             };
         }
     };
-    
-    private static final IRecordProcessorFactory SAMPLE_RECORD_PROCESSOR_FACTORY_V2 = 
+
+    private static final IRecordProcessorFactory SAMPLE_RECORD_PROCESSOR_FACTORY_V2 =
             new V1ToV2RecordProcessorFactoryAdapter(SAMPLE_RECORD_PROCESSOR_FACTORY);
 
 
@@ -232,6 +233,7 @@ public class WorkerTest {
                         parentShardPollIntervalMillis,
                         shardSyncIntervalMillis,
                         cleanupLeasesUponShardCompletion,
+                        ignoreUnexpectedChildShards,
                         checkpoint,
                         leaseCoordinator,
                         execService,
@@ -276,8 +278,9 @@ public class WorkerTest {
                 .thenReturn(secondCheckpoint);
 
         Worker worker = new Worker(stageName, streamletFactory, streamConfig, INITIAL_POSITION_LATEST,
-                parentShardPollIntervalMillis, shardSyncIntervalMillis, cleanupLeasesUponShardCompletion, checkpoint,
-                leaseCoordinator, execService, nullMetricsFactory, taskBackoffTimeMillis, failoverTimeMillis,
+                parentShardPollIntervalMillis, shardSyncIntervalMillis, cleanupLeasesUponShardCompletion,
+                ignoreUnexpectedChildShards, checkpoint, leaseCoordinator, execService, nullMetricsFactory,
+                taskBackoffTimeMillis, failoverTimeMillis,
                 KinesisClientLibConfiguration.DEFAULT_SKIP_SHARD_SYNC_AT_STARTUP_IF_LEASES_EXIST, shardPrioritization);
 
         Worker workerSpy = spy(worker);
@@ -336,6 +339,7 @@ public class WorkerTest {
                         parentShardPollIntervalMillis,
                         shardSyncIntervalMillis,
                         cleanupLeasesUponShardCompletion,
+                        ignoreUnexpectedChildShards,
                         checkpoint,
                         leaseCoordinator,
                         execService,
@@ -390,6 +394,7 @@ public class WorkerTest {
                         shardPollInterval,
                         shardSyncIntervalMillis,
                         cleanupLeasesUponShardCompletion,
+                        ignoreUnexpectedChildShards,
                         leaseCoordinator,
                         leaseCoordinator,
                         execService,
@@ -603,7 +608,7 @@ public class WorkerTest {
      * This test is testing the {@link Worker}'s shutdown behavior and by extension the behavior of
      * {@link ThreadPoolExecutor#shutdownNow()}. It depends on the thread pool sending an interrupt to the pool threads.
      * This behavior makes the test a bit racy, since we need to ensure a specific order of events.
-     * 
+     *
      * @throws Exception
      */
     @Test
@@ -744,8 +749,9 @@ public class WorkerTest {
 
         Worker worker = new Worker("testRequestShutdown", recordProcessorFactory, streamConfig,
                 INITIAL_POSITION_TRIM_HORIZON, parentShardPollIntervalMillis, shardSyncIntervalMillis,
-                cleanupLeasesUponShardCompletion, leaseCoordinator, leaseCoordinator, executorService, metricsFactory,
-                taskBackoffTimeMillis, failoverTimeMillis, false, shardPrioritization);
+                cleanupLeasesUponShardCompletion, ignoreUnexpectedChildShards, leaseCoordinator, leaseCoordinator,
+                executorService, metricsFactory, taskBackoffTimeMillis, failoverTimeMillis, false,
+                shardPrioritization);
 
         when(executorService.submit(Matchers.<Callable<TaskResult>> any()))
                 .thenAnswer(new ShutdownHandlingAnswer(taskFuture));
@@ -818,8 +824,8 @@ public class WorkerTest {
 
         Worker worker = new InjectableWorker("testRequestShutdown", recordProcessorFactory, streamConfig,
                 INITIAL_POSITION_TRIM_HORIZON, parentShardPollIntervalMillis, shardSyncIntervalMillis,
-                cleanupLeasesUponShardCompletion, leaseCoordinator, leaseCoordinator, executorService, metricsFactory,
-                taskBackoffTimeMillis, failoverTimeMillis, false, shardPrioritization) {
+                cleanupLeasesUponShardCompletion, ignoreUnexpectedChildShards, leaseCoordinator, leaseCoordinator,
+                executorService, metricsFactory, taskBackoffTimeMillis, failoverTimeMillis, false, shardPrioritization) {
             @Override
             void postConstruct() {
                 this.gracefuleShutdownStarted = true;
@@ -890,8 +896,8 @@ public class WorkerTest {
 
         Worker worker = new InjectableWorker("testRequestShutdown", recordProcessorFactory, streamConfig,
                 INITIAL_POSITION_TRIM_HORIZON, parentShardPollIntervalMillis, shardSyncIntervalMillis,
-                cleanupLeasesUponShardCompletion, leaseCoordinator, leaseCoordinator, executorService, metricsFactory,
-                taskBackoffTimeMillis, failoverTimeMillis, false, shardPrioritization) {
+                cleanupLeasesUponShardCompletion, ignoreUnexpectedChildShards, leaseCoordinator, leaseCoordinator,
+                executorService, metricsFactory, taskBackoffTimeMillis, failoverTimeMillis, false, shardPrioritization) {
             @Override
             void postConstruct() {
                 this.gracefulShutdownCoordinator = coordinator;
@@ -952,8 +958,9 @@ public class WorkerTest {
 
         Worker worker = new Worker("testRequestShutdown", recordProcessorFactory, streamConfig,
                 INITIAL_POSITION_TRIM_HORIZON, parentShardPollIntervalMillis, shardSyncIntervalMillis,
-                cleanupLeasesUponShardCompletion, leaseCoordinator, leaseCoordinator, executorService, metricsFactory,
-                taskBackoffTimeMillis, failoverTimeMillis, false, shardPrioritization);
+                cleanupLeasesUponShardCompletion, ignoreUnexpectedChildShards, leaseCoordinator, leaseCoordinator,
+                executorService, metricsFactory, taskBackoffTimeMillis, failoverTimeMillis, false,
+                shardPrioritization);
 
         when(executorService.submit(Matchers.<Callable<TaskResult>> any()))
                 .thenAnswer(new ShutdownHandlingAnswer(taskFuture));
@@ -1022,8 +1029,9 @@ public class WorkerTest {
 
         Worker worker = new Worker("testRequestShutdown", recordProcessorFactory, streamConfig,
                 INITIAL_POSITION_TRIM_HORIZON, parentShardPollIntervalMillis, shardSyncIntervalMillis,
-                cleanupLeasesUponShardCompletion, leaseCoordinator, leaseCoordinator, executorService, metricsFactory,
-                taskBackoffTimeMillis, failoverTimeMillis, false, shardPrioritization);
+                cleanupLeasesUponShardCompletion, ignoreUnexpectedChildShards, leaseCoordinator, leaseCoordinator,
+                executorService, metricsFactory, taskBackoffTimeMillis, failoverTimeMillis, false,
+                shardPrioritization);
 
         when(executorService.submit(Matchers.<Callable<TaskResult>> any()))
                 .thenAnswer(new ShutdownHandlingAnswer(taskFuture));
@@ -1123,8 +1131,9 @@ public class WorkerTest {
 
         Worker worker = new Worker("testRequestShutdown", recordProcessorFactory, streamConfig,
                 INITIAL_POSITION_TRIM_HORIZON, parentShardPollIntervalMillis, shardSyncIntervalMillis,
-                cleanupLeasesUponShardCompletion, leaseCoordinator, leaseCoordinator, executorService, metricsFactory,
-                taskBackoffTimeMillis, failoverTimeMillis, false, shardPrioritization);
+                cleanupLeasesUponShardCompletion, ignoreUnexpectedChildShards, leaseCoordinator, leaseCoordinator,
+                executorService, metricsFactory, taskBackoffTimeMillis, failoverTimeMillis, false,
+                shardPrioritization);
 
         when(executorService.submit(Matchers.<Callable<TaskResult>> any()))
                 .thenAnswer(new ShutdownHandlingAnswer(taskFuture));
@@ -1228,8 +1237,9 @@ public class WorkerTest {
 
         Worker worker = new Worker("testRequestShutdown", recordProcessorFactory, streamConfig,
                 INITIAL_POSITION_TRIM_HORIZON, parentShardPollIntervalMillis, shardSyncIntervalMillis,
-                cleanupLeasesUponShardCompletion, leaseCoordinator, leaseCoordinator, executorService, metricsFactory,
-                taskBackoffTimeMillis, failoverTimeMillis, false, shardPrioritization);
+                cleanupLeasesUponShardCompletion, ignoreUnexpectedChildShards, leaseCoordinator, leaseCoordinator,
+                executorService, metricsFactory, taskBackoffTimeMillis, failoverTimeMillis, false,
+                shardPrioritization);
 
         when(executorService.submit(Matchers.<Callable<TaskResult>> any()))
                 .thenAnswer(new ShutdownHandlingAnswer(taskFuture));
@@ -1300,8 +1310,9 @@ public class WorkerTest {
 
         Worker worker = new Worker("testRequestShutdown", recordProcessorFactory, streamConfig,
                 INITIAL_POSITION_TRIM_HORIZON, parentShardPollIntervalMillis, shardSyncIntervalMillis,
-                cleanupLeasesUponShardCompletion, leaseCoordinator, leaseCoordinator, executorService, metricsFactory,
-                taskBackoffTimeMillis, failoverTimeMillis, false, shardPrioritization);
+                cleanupLeasesUponShardCompletion, ignoreUnexpectedChildShards, leaseCoordinator, leaseCoordinator,
+                executorService, metricsFactory, taskBackoffTimeMillis, failoverTimeMillis, false,
+                shardPrioritization);
 
         when(executorService.submit(Matchers.<Callable<TaskResult>> any()))
                 .thenAnswer(new ShutdownHandlingAnswer(taskFuture));
@@ -1338,14 +1349,15 @@ public class WorkerTest {
         InjectableWorker(String applicationName, IRecordProcessorFactory recordProcessorFactory,
                 StreamConfig streamConfig, InitialPositionInStreamExtended initialPositionInStream,
                 long parentShardPollIntervalMillis, long shardSyncIdleTimeMillis,
-                boolean cleanupLeasesUponShardCompletion, ICheckpoint checkpoint,
+                boolean cleanupLeasesUponShardCompletion, boolean ignoreUnexpectedChildShards, ICheckpoint checkpoint,
                 KinesisClientLibLeaseCoordinator leaseCoordinator, ExecutorService execService,
                 IMetricsFactory metricsFactory, long taskBackoffTimeMillis, long failoverTimeMillis,
                 boolean skipShardSyncAtWorkerInitializationIfLeasesExist, ShardPrioritization shardPrioritization) {
             super(applicationName, recordProcessorFactory, streamConfig, initialPositionInStream,
                     parentShardPollIntervalMillis, shardSyncIdleTimeMillis, cleanupLeasesUponShardCompletion,
-                    checkpoint, leaseCoordinator, execService, metricsFactory, taskBackoffTimeMillis,
-                    failoverTimeMillis, skipShardSyncAtWorkerInitializationIfLeasesExist, shardPrioritization);
+                    ignoreUnexpectedChildShards, checkpoint, leaseCoordinator, execService, metricsFactory,
+                    taskBackoffTimeMillis, failoverTimeMillis, skipShardSyncAtWorkerInitializationIfLeasesExist,
+                    shardPrioritization);
             postConstruct();
         }
 
@@ -1657,6 +1669,7 @@ public class WorkerTest {
                         parentShardPollIntervalMillis,
                         shardSyncIntervalMillis,
                         cleanupLeasesUponShardCompletion,
+                        ignoreUnexpectedChildShards,
                         leaseCoordinator,
                         leaseCoordinator,
                         executorService,
@@ -1665,7 +1678,7 @@ public class WorkerTest {
                         failoverTimeMillis,
                         KinesisClientLibConfiguration.DEFAULT_SKIP_SHARD_SYNC_AT_STARTUP_IF_LEASES_EXIST,
                         shardPrioritization);
-        
+
         WorkerThread workerThread = new WorkerThread(worker);
         workerThread.start();
         return workerThread;