@Test public final void testCallWhenApplicationDoesNotCheckpoint() throws Exception { when(recordProcessorCheckpointer.lastCheckpointValue()).thenReturn(new ExtendedSequenceNumber("3298")); Lease heldLease = LeaseHelper.createLease("shardId-0", "leaseOwner", Collections.singleton("parentShardId"), Collections.emptyList(), ExtendedSequenceNumber.LATEST); when(leaseCoordinator.getCurrentlyHeldLease("shardId-0")).thenReturn(heldLease); when(leaseRefresher.getLease("shardId-0")).thenReturn(heldLease); when(leaseCoordinator.leaseRefresher()).thenReturn(leaseRefresher); when(leaseCoordinator.updateLease(Matchers.any(Lease.class), Matchers.any(UUID.class), Matchers.anyString(), Matchers.anyString())).thenReturn(true); final TaskResult result = task.call(); assertNotNull(result.getException()); assertTrue(result.getException() instanceof CustomerApplicationException); }
@Override public TaskResult call() { recordProcessorCheckpointer.checkpointer().operation(SHUTDOWN_TASK_OPERATION); final MetricsScope scope = MetricsUtil.createMetricsWithOperation(metricsFactory, SHUTDOWN_TASK_OPERATION); Exception exception; try { try { log.debug("Invoking shutdown() for shard {} with childShards {}, concurrencyToken {}. Shutdown reason: {}", leaseKeyProvider.apply(shardInfo), childShards, shardInfo.concurrencyToken(), reason); final long startTime = System.currentTimeMillis(); final Lease currentShardLease = leaseCoordinator.getCurrentlyHeldLease(leaseKeyProvider.apply(shardInfo)); final Runnable leaseLostAction = () -> shardRecordProcessor.leaseLost(LeaseLostInput.builder().build()); if (reason == ShutdownReason.SHARD_END) { try { takeShardEndAction(currentShardLease, scope, startTime); } catch (InvalidStateException e) { log.warn("Lease {}: Invalid state encountered while shutting down shardConsumer with SHARD_END reason. " + "Dropping the lease and shutting down shardConsumer using LEASE_LOST reason. ", leaseKeyProvider.apply(shardInfo), e); dropLease(currentShardLease); throwOnApplicationException(leaseLostAction, scope, startTime); } } else { throwOnApplicationException(leaseLostAction, scope, startTime); } log.debug("Shutting down retrieval strategy for shard {}.", leaseKeyProvider.apply(shardInfo)); recordsPublisher.shutdown(); log.debug("Record processor completed shutdown() for shard {}", leaseKeyProvider.apply(shardInfo)); return new TaskResult(null); } catch (Exception e) { if (e instanceof CustomerApplicationException) { log.error("Shard {}: Application exception. ", leaseKeyProvider.apply(shardInfo), e); } else { log.error("Shard {}: Caught exception: ", leaseKeyProvider.apply(shardInfo), e); } exception = e; try { Thread.sleep(this.backoffTimeMillis); } catch (InterruptedException ie) { log.debug("Shard {}: Interrupted sleep", leaseKeyProvider.apply(shardInfo), ie); } } } finally { MetricsUtil.endScope(scope); } return new TaskResult(exception); }
ShutdownTask implements ConsumerTask { @Override public TaskResult call() { recordProcessorCheckpointer.checkpointer().operation(SHUTDOWN_TASK_OPERATION); final MetricsScope scope = MetricsUtil.createMetricsWithOperation(metricsFactory, SHUTDOWN_TASK_OPERATION); Exception exception; try { try { log.debug("Invoking shutdown() for shard {} with childShards {}, concurrencyToken {}. Shutdown reason: {}", leaseKeyProvider.apply(shardInfo), childShards, shardInfo.concurrencyToken(), reason); final long startTime = System.currentTimeMillis(); final Lease currentShardLease = leaseCoordinator.getCurrentlyHeldLease(leaseKeyProvider.apply(shardInfo)); final Runnable leaseLostAction = () -> shardRecordProcessor.leaseLost(LeaseLostInput.builder().build()); if (reason == ShutdownReason.SHARD_END) { try { takeShardEndAction(currentShardLease, scope, startTime); } catch (InvalidStateException e) { log.warn("Lease {}: Invalid state encountered while shutting down shardConsumer with SHARD_END reason. " + "Dropping the lease and shutting down shardConsumer using LEASE_LOST reason. ", leaseKeyProvider.apply(shardInfo), e); dropLease(currentShardLease); throwOnApplicationException(leaseLostAction, scope, startTime); } } else { throwOnApplicationException(leaseLostAction, scope, startTime); } log.debug("Shutting down retrieval strategy for shard {}.", leaseKeyProvider.apply(shardInfo)); recordsPublisher.shutdown(); log.debug("Record processor completed shutdown() for shard {}", leaseKeyProvider.apply(shardInfo)); return new TaskResult(null); } catch (Exception e) { if (e instanceof CustomerApplicationException) { log.error("Shard {}: Application exception. ", leaseKeyProvider.apply(shardInfo), e); } else { log.error("Shard {}: Caught exception: ", leaseKeyProvider.apply(shardInfo), e); } exception = e; try { Thread.sleep(this.backoffTimeMillis); } catch (InterruptedException ie) { log.debug("Shard {}: Interrupted sleep", leaseKeyProvider.apply(shardInfo), ie); } } } finally { MetricsUtil.endScope(scope); } return new TaskResult(exception); } }
ShutdownTask implements ConsumerTask { @Override public TaskResult call() { recordProcessorCheckpointer.checkpointer().operation(SHUTDOWN_TASK_OPERATION); final MetricsScope scope = MetricsUtil.createMetricsWithOperation(metricsFactory, SHUTDOWN_TASK_OPERATION); Exception exception; try { try { log.debug("Invoking shutdown() for shard {} with childShards {}, concurrencyToken {}. Shutdown reason: {}", leaseKeyProvider.apply(shardInfo), childShards, shardInfo.concurrencyToken(), reason); final long startTime = System.currentTimeMillis(); final Lease currentShardLease = leaseCoordinator.getCurrentlyHeldLease(leaseKeyProvider.apply(shardInfo)); final Runnable leaseLostAction = () -> shardRecordProcessor.leaseLost(LeaseLostInput.builder().build()); if (reason == ShutdownReason.SHARD_END) { try { takeShardEndAction(currentShardLease, scope, startTime); } catch (InvalidStateException e) { log.warn("Lease {}: Invalid state encountered while shutting down shardConsumer with SHARD_END reason. " + "Dropping the lease and shutting down shardConsumer using LEASE_LOST reason. ", leaseKeyProvider.apply(shardInfo), e); dropLease(currentShardLease); throwOnApplicationException(leaseLostAction, scope, startTime); } } else { throwOnApplicationException(leaseLostAction, scope, startTime); } log.debug("Shutting down retrieval strategy for shard {}.", leaseKeyProvider.apply(shardInfo)); recordsPublisher.shutdown(); log.debug("Record processor completed shutdown() for shard {}", leaseKeyProvider.apply(shardInfo)); return new TaskResult(null); } catch (Exception e) { if (e instanceof CustomerApplicationException) { log.error("Shard {}: Application exception. ", leaseKeyProvider.apply(shardInfo), e); } else { log.error("Shard {}: Caught exception: ", leaseKeyProvider.apply(shardInfo), e); } exception = e; try { Thread.sleep(this.backoffTimeMillis); } catch (InterruptedException ie) { log.debug("Shard {}: Interrupted sleep", leaseKeyProvider.apply(shardInfo), ie); } } } finally { MetricsUtil.endScope(scope); } return new TaskResult(exception); }  }
ShutdownTask implements ConsumerTask { @Override public TaskResult call() { recordProcessorCheckpointer.checkpointer().operation(SHUTDOWN_TASK_OPERATION); final MetricsScope scope = MetricsUtil.createMetricsWithOperation(metricsFactory, SHUTDOWN_TASK_OPERATION); Exception exception; try { try { log.debug("Invoking shutdown() for shard {} with childShards {}, concurrencyToken {}. Shutdown reason: {}", leaseKeyProvider.apply(shardInfo), childShards, shardInfo.concurrencyToken(), reason); final long startTime = System.currentTimeMillis(); final Lease currentShardLease = leaseCoordinator.getCurrentlyHeldLease(leaseKeyProvider.apply(shardInfo)); final Runnable leaseLostAction = () -> shardRecordProcessor.leaseLost(LeaseLostInput.builder().build()); if (reason == ShutdownReason.SHARD_END) { try { takeShardEndAction(currentShardLease, scope, startTime); } catch (InvalidStateException e) { log.warn("Lease {}: Invalid state encountered while shutting down shardConsumer with SHARD_END reason. " + "Dropping the lease and shutting down shardConsumer using LEASE_LOST reason. ", leaseKeyProvider.apply(shardInfo), e); dropLease(currentShardLease); throwOnApplicationException(leaseLostAction, scope, startTime); } } else { throwOnApplicationException(leaseLostAction, scope, startTime); } log.debug("Shutting down retrieval strategy for shard {}.", leaseKeyProvider.apply(shardInfo)); recordsPublisher.shutdown(); log.debug("Record processor completed shutdown() for shard {}", leaseKeyProvider.apply(shardInfo)); return new TaskResult(null); } catch (Exception e) { if (e instanceof CustomerApplicationException) { log.error("Shard {}: Application exception. ", leaseKeyProvider.apply(shardInfo), e); } else { log.error("Shard {}: Caught exception: ", leaseKeyProvider.apply(shardInfo), e); } exception = e; try { Thread.sleep(this.backoffTimeMillis); } catch (InterruptedException ie) { log.debug("Shard {}: Interrupted sleep", leaseKeyProvider.apply(shardInfo), ie); } } } finally { MetricsUtil.endScope(scope); } return new TaskResult(exception); }  @Override TaskResult call(); @Override TaskType taskType(); @VisibleForTesting ShutdownReason getReason(); }
ShutdownTask implements ConsumerTask { @Override public TaskResult call() { recordProcessorCheckpointer.checkpointer().operation(SHUTDOWN_TASK_OPERATION); final MetricsScope scope = MetricsUtil.createMetricsWithOperation(metricsFactory, SHUTDOWN_TASK_OPERATION); Exception exception; try { try { log.debug("Invoking shutdown() for shard {} with childShards {}, concurrencyToken {}. Shutdown reason: {}", leaseKeyProvider.apply(shardInfo), childShards, shardInfo.concurrencyToken(), reason); final long startTime = System.currentTimeMillis(); final Lease currentShardLease = leaseCoordinator.getCurrentlyHeldLease(leaseKeyProvider.apply(shardInfo)); final Runnable leaseLostAction = () -> shardRecordProcessor.leaseLost(LeaseLostInput.builder().build()); if (reason == ShutdownReason.SHARD_END) { try { takeShardEndAction(currentShardLease, scope, startTime); } catch (InvalidStateException e) { log.warn("Lease {}: Invalid state encountered while shutting down shardConsumer with SHARD_END reason. " + "Dropping the lease and shutting down shardConsumer using LEASE_LOST reason. ", leaseKeyProvider.apply(shardInfo), e); dropLease(currentShardLease); throwOnApplicationException(leaseLostAction, scope, startTime); } } else { throwOnApplicationException(leaseLostAction, scope, startTime); } log.debug("Shutting down retrieval strategy for shard {}.", leaseKeyProvider.apply(shardInfo)); recordsPublisher.shutdown(); log.debug("Record processor completed shutdown() for shard {}", leaseKeyProvider.apply(shardInfo)); return new TaskResult(null); } catch (Exception e) { if (e instanceof CustomerApplicationException) { log.error("Shard {}: Application exception. ", leaseKeyProvider.apply(shardInfo), e); } else { log.error("Shard {}: Caught exception: ", leaseKeyProvider.apply(shardInfo), e); } exception = e; try { Thread.sleep(this.backoffTimeMillis); } catch (InterruptedException ie) { log.debug("Shard {}: Interrupted sleep", leaseKeyProvider.apply(shardInfo), ie); } } } finally { MetricsUtil.endScope(scope); } return new TaskResult(exception); }  @Override TaskResult call(); @Override TaskType taskType(); @VisibleForTesting ShutdownReason getReason();  }