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

LeaseCleanupManager change

Browse source
This commit is contained in:
Chunxue Yang 2020-07-28 13:11:57 -07:00
parent c3b41c3b55
commit f713017528
3 changed files with 22 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/Worker.java
View file

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

17
src/main/java/com/amazonaws/services/kinesis/leases/impl/LeaseCleanupManager.java
View file

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

11
src/test/java/com/amazonaws/services/kinesis/leases/impl/LeaseCleanupManagerTest.java
View file

@ -26,6 +26,7 @@ import com.amazonaws.services.kinesis.model.ChildShard;
import com.amazonaws.services.kinesis.model.GetRecordsResult;
import com.amazonaws.services.kinesis.model.ResourceNotFoundException;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@ -82,6 +83,16 @@ public class LeaseCleanupManagerTest {
cleanupLeasesOfCompletedShards, leaseCleanupIntervalMillis, completedLeaseCleanupIntervalMillis, garbageLeaseCleanupIntervalMillis, maxRecords);
}
/**
* 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
* shard case.