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

Merge pull request #85 from joshua-kim/ltr_1

Browse source 
Fixing bug where isRunning is not set
This commit is contained in:
ashwing 2020-07-27 16:42:51 -07:00 committed by GitHub
commit ca661bf30d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 22 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -348,12 +348,7 @@ public class Scheduler implements Runnable {
log.info("Skipping shard sync per configuration setting (and lease table is not empty)"); log.info("Skipping shard sync per configuration setting (and lease table is not empty)");
} }
if (!leaseCleanupManager.isRunning()) {
log.info("Starting LeaseCleanupManager.");
leaseCleanupManager.start(); leaseCleanupManager.start();
} else {
log.info("LeaseCleanupManager is already running. No need to start it");
}
// If we reach this point, then we either skipped the lease sync or did not have any exception // If we reach this point, then we either skipped the lease sync or did not have any exception
// for any of the shard sync in the previous attempt. // for any of the shard sync in the previous attempt.

12
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseCleanupManager.java
View file

@ -81,12 +81,16 @@ public class LeaseCleanupManager {
* {@link LeaseCleanupManager#leaseCleanupIntervalMillis} * {@link LeaseCleanupManager#leaseCleanupIntervalMillis}
*/ */
public void start() { public void start() {
log.debug("Starting lease cleanup thread."); if (!isRunning) {
completedLeaseStopwatch.start(); log.info("Starting lease cleanup thread.");
garbageLeaseStopwatch.start(); completedLeaseStopwatch.reset().start();
garbageLeaseStopwatch.reset().start();
deletionThreadPool.scheduleAtFixedRate(new LeaseCleanupThread(), INITIAL_DELAY, leaseCleanupIntervalMillis, deletionThreadPool.scheduleAtFixedRate(new LeaseCleanupThread(), INITIAL_DELAY, leaseCleanupIntervalMillis,
TimeUnit.MILLISECONDS); TimeUnit.MILLISECONDS);
isRunning = true;
} else {
log.info("Lease cleanup thread already running, no need to start.");
}
} }
/** /**

11
amazon-kinesis-client/src/test/java/software/amazon/kinesis/leases/LeaseCleanupManagerTest.java
View file

@ -15,6 +15,7 @@
package software.amazon.kinesis.leases; package software.amazon.kinesis.leases;
import org.junit.Assert;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.junit.runner.RunWith; import org.junit.runner.RunWith;
@ -83,6 +84,16 @@ public class LeaseCleanupManagerTest {
when(leaseCoordinator.updateLease(any(Lease.class), any(UUID.class), any(String.class), any(String.class))).thenReturn(true); when(leaseCoordinator.updateLease(any(Lease.class), any(UUID.class), any(String.class), any(String.class))).thenReturn(true);
} }
/**
* Tests subsequent calls to start {@link LeaseCleanupManager}.
*/
@Test
public final void testSubsequentStarts() {
leaseCleanupManager.start();
Assert.assertTrue(leaseCleanupManager.isRunning());
leaseCleanupManager.start();
}
/** /**
* Tests that when both child shard leases are present, we are able to delete the parent shard for the completed * Tests that when both child shard leases are present, we are able to delete the parent shard for the completed
* shard case. * shard case.