You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sling.apache.org by cs...@apache.org on 2020/07/02 15:04:57 UTC
[sling-org-apache-sling-distribution-journal] branch SLING-9563
created (now bf0a1da)
This is an automated email from the ASF dual-hosted git repository.
cschneider pushed a change to branch SLING-9563
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git.
at bf0a1da SLING-9563 - Use InterruptedException to exit flow
This branch includes the following new commits:
new bf0a1da SLING-9563 - Use InterruptedException to exit flow
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[sling-org-apache-sling-distribution-journal] 01/01: SLING-9563 -
Use InterruptedException to exit flow
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
cschneider pushed a commit to branch SLING-9563
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git
commit bf0a1da2a59ec9ef6a6ced04f16b4184896c660a
Author: Christian Schneider <cs...@adobe.com>
AuthorDate: Thu Jul 2 17:03:13 2020 +0200
SLING-9563 - Use InterruptedException to exit flow
---
.../impl/subscriber/DistributionSubscriber.java | 74 ++++++++++------------
1 file changed, 34 insertions(+), 40 deletions(-)
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/DistributionSubscriber.java b/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/DistributionSubscriber.java
index 6b7ddd1..0d42d6b 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/DistributionSubscriber.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/subscriber/DistributionSubscriber.java
@@ -273,67 +273,61 @@ public class DistributionSubscriber {
private void processQueue() {
LOG.info("Started Queue processor");
while (running) {
- fetchAndProcessQueueItem();
+ try {
+ fetchAndProcessQueueItem();
+ } catch (PreConditionTimeoutException e) {
+ // Precondition timed out. We only log this on info level as it is no error
+ LOG.info(e.getMessage());
+ delay(RETRY_DELAY);
+ } catch (InterruptedException e) {
+ LOG.debug(e.getMessage());
+ } catch (Exception e) {
+ // Catch all to prevent processing from stopping
+ LOG.error("Error processing queue item", e);
+ delay(RETRY_DELAY);
+ }
}
LOG.info("Stopped Queue processor");
}
- private void fetchAndProcessQueueItem() {
- try {
-
- if (! blockingSendStoredStatus()) {
- return;
- }
-
- Optional<FullMessage<PackageMessage>> item = blockingPeekQueueItem();
- if (!item.isPresent()) {
- return;
- }
-
- try (Timer.Context context = distributionMetricsService.getProcessQueueItemDuration().time()) {
- processQueueItem(item.get());
- messageBuffer.remove();
- distributionMetricsService.getItemsBufferSize().decrement();
- }
-
- } catch (PreConditionTimeoutException e) {
- // Precondition timed out. We only log this on info level as it is no error
- LOG.info(e.getMessage());
- delay(RETRY_DELAY);
- } catch (IllegalStateException e) {
- throw e;
- } catch (Exception e) {
- // Catch all to prevent processing from stopping
- LOG.error("Error processing queue item", e);
- delay(RETRY_DELAY);
+ private void fetchAndProcessQueueItem() throws InterruptedException, IOException, LoginException, DistributionException {
+ blockingSendStoredStatus();
+ FullMessage<PackageMessage> item = blockingPeekQueueItem();
+ try (Timer.Context context = distributionMetricsService.getProcessQueueItemDuration().time()) {
+ processQueueItem(item);
+ messageBuffer.remove();
+ distributionMetricsService.getItemsBufferSize().decrement();
}
}
/**
* Send status stored in a previous run if exists
- *
- * @return {@code true} if the status has been sent ;
- * {@code false} otherwise.
+ * @throws InterruptedException in case of shutdown
+ * @throws IOException from time.close. Should not happen
*/
- private boolean blockingSendStoredStatus() {
+ private void blockingSendStoredStatus() throws InterruptedException, IOException {
try (Timer.Context context = distributionMetricsService.getSendStoredStatusDuration().time()) {
- for (int retry = 0 ; running && ! bookKeeper.sendStoredStatus(retry) ; retry++);
- } catch (IOException e) {
- LOG.warn("Error in timer close", e);
+ int retry = 0;
+ while (running) {
+ if (bookKeeper.sendStoredStatus(retry)) {
+ return;
+ }
+ retry++;
+ }
}
- return running;
+ throw new InterruptedException("Shutting down");
}
- private Optional<FullMessage<PackageMessage>> blockingPeekQueueItem() throws InterruptedException {
+ private FullMessage<PackageMessage> blockingPeekQueueItem() throws InterruptedException {
while (running) {
FullMessage<PackageMessage> message = messageBuffer.peek();
if (message != null) {
- return Optional.of(message);
+ return message;
} else {
Thread.sleep(QUEUE_FETCH_DELAY);
}
}
- return Optional.empty();
+ throw new InterruptedException("Shutting down");
}
private void processQueueItem(FullMessage<PackageMessage> item) throws PersistenceException, LoginException, DistributionException {