Run multiple instance of scheduler on one JVM #395

xiaoyu-meng-mxy · 2018-09-07T02:35:19Z

Description of changes:
Enable KCL V2 to run multiple instances of the Scheduler on a single JVM.
Multiple KCL schedulers should be able to run with different event source mappings of different customers in the same JVM.

Why?
Lambda event bridge application handles a high number of event source mappings of the customers. Running multiple KCL schedulers on the same instance and on the same JVM reduces the number of hosts required process the streams.

Solution:
There are static synchronized functions in KCL and they are not suitable for this scenario. They have to be modified to be synchronized per event source mapping and not per class.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

pfifer

A couple of small changes

pfifer · 2018-09-12T16:49:37Z

amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ShardSyncer.java

 * It will create new leases/activities when it discovers new Kinesis shards (bootstrap/resharding).
 * It deletes leases for shards that have been trimmed from Kinesis, or if we've completed processing it
 * and begun processing it's child shards.
 */
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
+@NoArgsConstructor(access = AccessLevel.PUBLIC)


You shouldn't need to add this.

pfifer · 2018-09-12T16:57:08Z

amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ShardSyncTaskManager.java

+    public ShardSyncTaskManager(ShardDetector shardDetector, LeaseRefresher leaseRefresher, InitialPositionInStreamExtended initialPositionInStream,
+                                boolean cleanupLeasesUponShardCompletion, boolean ignoreUnexpectedChildShards, long shardSyncIdleTimeMillis,
+                                ExecutorService executorService, MetricsFactory metricsFactory) {
+        this(shardDetector, leaseRefresher, initialPositionInStream, cleanupLeasesUponShardCompletion, ignoreUnexpectedChildShards, shardSyncIdleTimeMillis, executorService, metricsFactory, new ShardSyncer());


You should not depend on the ShardSyncTaskManager creating the ShardSyncer, instead handle the creation in DynamoDBLeaseManagementFactory.

…d LeaseManagementConfig

sahilpalvia · 2018-09-17T20:08:19Z

amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManagementConfig.java

@@ -220,6 +220,15 @@ public LeaseManagementConfig metricsFactory(final MetricsFactory metricsFactory)
        }
    };

+    private ShardSyncer shardSyncer;


No need to create the getter

private ShardSyncer shardSyncer = new ShardSyncer();

This should work.

yes you are right

sahilpalvia · 2018-09-17T20:09:47Z

amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ShardSyncTaskManager.java

+    @NonNull
+    private final ShardSyncer shardSyncer;
+
+    public ShardSyncTaskManager(ShardDetector shardDetector, LeaseRefresher leaseRefresher, InitialPositionInStreamExtended initialPositionInStream,


Could you change the order of MetricsFactory and ShardSyncer. We have maintained MetricsFactory as the last parameter of the constructors.

sahilpalvia · 2018-09-17T20:10:50Z

amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ShardSyncer.java

@@ -71,7 +68,7 @@
     * @throws KinesisClientLibIOException
     */
    // CHECKSTYLE:OFF CyclomaticComplexity
-    public static synchronized void checkAndCreateLeasesForNewShards(@NonNull final ShardDetector shardDetector,
+    public synchronized void checkAndCreateLeasesForNewShards(@NonNull final ShardDetector shardDetector,


This seems like a breaking change. A better way would be to deprecate the existing method and creating a new non-static method.

You are right, modified.

sahilpalvia · 2018-09-17T21:33:06Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/coordinator/SchedulerTest.java

+     * worker and checks if the second worker can start syncing shards.
+     */
+    @Test(timeout = 30000L)
+    public final void testWorkersCanSyncShardsInParallel() {


Avoid multithreaded testing in unit tests. Testing a single instance should be fine for unit testing.

OK, will remove this unit test.

…e the order for metricsFactory and HierarchichalShardSyncer in ShardConsumerArgument

sahilpalvia

Looking good, just a few more minor changes

sahilpalvia · 2018-09-24T18:25:58Z

amazon-kinesis-client/src/main/java/software/amazon/kinesis/coordinator/Scheduler.java

@@ -236,7 +239,8 @@ private void initialize() {
                if (!skipShardSyncAtWorkerInitializationIfLeasesExist || leaseRefresher.isLeaseTableEmpty()) {
                    log.info("Syncing Kinesis shard info");
                    ShardSyncTask shardSyncTask = new ShardSyncTask(shardDetector, leaseRefresher, initialPosition,
-                            cleanupLeasesUponShardCompletion, ignoreUnexpetedChildShards, 0L, metricsFactory);
+                            cleanupLeasesUponShardCompletion, ignoreUnexpetedChildShards, 0L,
+                            hierarchichalShardSyncer, metricsFactory);


Misspelled Hierarchical.

sahilpalvia · 2018-09-24T18:30:45Z

amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ShardSyncTaskManager.java

    private final MetricsFactory metricsFactory;

+    public ShardSyncTaskManager(ShardDetector shardDetector, LeaseRefresher leaseRefresher, InitialPositionInStreamExtended initialPositionInStream,


Could you create the deprecated constructor too, with the default being HierarchicalShardSyncer.

sahilpalvia · 2018-09-24T18:31:29Z

amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/ShardSyncTaskManager.java

@@ -81,7 +97,7 @@ private synchronized boolean checkAndSubmitNextTask() {
                                    initialPositionInStream,
                                    cleanupLeasesUponShardCompletion,
                                    ignoreUnexpectedChildShards,
-                                    shardSyncIdleTimeMillis,
+                                    shardSyncIdleTimeMillis, hierarchichalShardSyncer,


Could you follow the formatting style for the file and include the parameter on a new line.

sahilpalvia · 2018-09-24T20:22:19Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/lifecycle/ShutdownTaskTest.java

@@ -54,6 +55,7 @@
            InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.TRIM_HORIZON);
    private static final ShutdownReason TERMINATE_SHUTDOWN_REASON = ShutdownReason.SHARD_END;
    private static final MetricsFactory NULL_METRICS_FACTORY = new NullMetricsFactory();
+    private static final HierarchichalShardSyncer SHARD_SYNCER = new HierarchichalShardSyncer();


Does it need to be an instance of the ShardSyncer? Or can it be mocked?

sahilpalvia · 2018-09-24T20:23:09Z

amazon-kinesis-client/src/test/java/software/amazon/kinesis/lifecycle/ConsumerStatesTest.java

@@ -109,7 +110,8 @@ public void setup() {
                taskBackoffTimeMillis, skipShardSyncAtWorkerInitializationIfLeasesExist,
                listShardsBackoffTimeInMillis, maxListShardsRetryAttempts,
                shouldCallProcessRecordsEvenForEmptyRecordList, idleTimeInMillis, INITIAL_POSITION_IN_STREAM,
-                cleanupLeasesOfCompletedShards, ignoreUnexpectedChildShards, shardDetector, metricsFactory, new AggregatorUtil());
+                cleanupLeasesOfCompletedShards, ignoreUnexpectedChildShards, shardDetector, new AggregatorUtil(),
+                new HierarchichalShardSyncer(), metricsFactory);


Does it need to be a new instance of the ShardSyncer? Or can it be a mock object?

before this change, you used static methods in ShardSyncer.class and you never mock it. That's why I think we need to create a new instance.

Changed to use mock

sahilpalvia · 2018-09-24T20:24:38Z

...nt/src/main/java/software/amazon/kinesis/leases/dynamodb/DynamoDBLeaseManagementFactory.java

+     * @param initialLeaseTableWriteCapacity
+     * @param hierarchichalShardSyncer
+     */
+    public DynamoDBLeaseManagementFactory(final KinesisAsyncClient kinesisClient, final String streamName,


Could you make it use the formatter on this change.

sahilpalvia

Looking good. Thanks for providing this.

pfifer · 2018-10-09T14:40:09Z

Thanks for the change, and sorry about the delay.

One more thing can you please pull resolve the conflicts, and update the PR.

Run multiple instance of scheduler on one JVM

ea999c5

pfifer added the v2.x Issues related to the 2.x version label Sep 7, 2018

pfifer suggested changes Sep 12, 2018

View reviewed changes

handling creation of shardSyncer in DynamoDBLeaseManagementFactory an…

ab29cd2

…d LeaseManagementConfig

sahilpalvia suggested changes Sep 17, 2018

View reviewed changes

sahilpalvia reviewed Sep 17, 2018

View reviewed changes

Meng added 2 commits September 18, 2018 14:35

remove multi-threading unit test and do some small refactorings

9fa3fe9

refectoring

b598d44

sahilpalvia mentioned this pull request Sep 21, 2018

KCL consumer does not take all shards while consuming multiple streams #415

Closed

Meng added 2 commits September 21, 2018 16:28

deprecate ShardSyncer and use HierarchichalShardSyncer instead; chang…

474209c

…e the order for metricsFactory and HierarchichalShardSyncer in ShardConsumerArgument

fix merge conflict

0a6a333

sahilpalvia suggested changes Sep 24, 2018

View reviewed changes

Meng added 4 commits September 24, 2018 14:55

fix typos and use mock object of shardSyncer

300e480

delete improper comments

a305e34

fix comments

227c96f

remove duplicated comments

4b60aab

sahilpalvia approved these changes Sep 24, 2018

View reviewed changes

pfifer approved these changes Oct 9, 2018

View reviewed changes

pfifer added this to the v2.0.4 milestone Oct 9, 2018

fix merge from master

5073d25

sahilpalvia merged commit 14c6829 into awslabs:master Oct 10, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Run multiple instance of scheduler on one JVM #395

Run multiple instance of scheduler on one JVM #395

xiaoyu-meng-mxy commented Sep 7, 2018

pfifer left a comment

pfifer Sep 12, 2018

xiaoyu-meng-mxy Sep 13, 2018

pfifer Sep 12, 2018

xiaoyu-meng-mxy Sep 13, 2018

sahilpalvia Sep 17, 2018

xiaoyu-meng-mxy Sep 18, 2018

sahilpalvia Sep 17, 2018

xiaoyu-meng-mxy Sep 18, 2018 •

edited

Loading

sahilpalvia Sep 17, 2018

xiaoyu-meng-mxy Sep 18, 2018

sahilpalvia Sep 17, 2018

xiaoyu-meng-mxy Sep 18, 2018

sahilpalvia left a comment

sahilpalvia Sep 24, 2018

xiaoyu-meng-mxy Sep 24, 2018

sahilpalvia Sep 24, 2018

xiaoyu-meng-mxy Sep 24, 2018

sahilpalvia Sep 24, 2018

xiaoyu-meng-mxy Sep 24, 2018

sahilpalvia Sep 24, 2018

xiaoyu-meng-mxy Sep 24, 2018

sahilpalvia Sep 24, 2018

xiaoyu-meng-mxy Sep 24, 2018

xiaoyu-meng-mxy Sep 24, 2018

sahilpalvia Sep 24, 2018 •

edited

Loading

xiaoyu-meng-mxy Sep 24, 2018

sahilpalvia left a comment

pfifer commented Oct 9, 2018

		private final MetricsFactory metricsFactory;

		public ShardSyncTaskManager(ShardDetector shardDetector, LeaseRefresher leaseRefresher, InitialPositionInStreamExtended initialPositionInStream,

Run multiple instance of scheduler on one JVM #395

Run multiple instance of scheduler on one JVM #395

Conversation

xiaoyu-meng-mxy commented Sep 7, 2018

pfifer left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xiaoyu-meng-mxy Sep 18, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sahilpalvia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sahilpalvia Sep 24, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sahilpalvia left a comment

Choose a reason for hiding this comment

pfifer commented Oct 9, 2018

xiaoyu-meng-mxy Sep 18, 2018 •

edited

Loading

sahilpalvia Sep 24, 2018 •

edited

Loading