This is an automated email from the ASF dual-hosted git repository. cschneider pushed a commit to branch SLING-8859 in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git
commit d2ae52a483f48680cab2a195cc570b0717a0f0d0 Author: Christian Schneider <[email protected]> AuthorDate: Tue Nov 26 12:08:42 2019 +0100 SLING-8859 - Fix double JournalAvailable service and sync issues --- .../journal/impl/queue/impl/PubQueueCache.java | 17 +++++--- .../impl/shared/JournalAvailableChecker.java | 37 ++++++---------- .../impl/shared/JournalAvailableServiceMarker.java | 50 ++++++++++++++++++++++ 3 files changed, 73 insertions(+), 31 deletions(-) diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCache.java b/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCache.java index 89a001f..9a94691 100644 --- a/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCache.java +++ b/src/main/java/org/apache/sling/distribution/journal/impl/queue/impl/PubQueueCache.java @@ -85,6 +85,7 @@ public class PubQueueCache { * Interval in millisecond between two seeding messages. */ private static final long SEEDING_DELAY_MS = 1000; + private static final long SEEDING_ERROR_DELAY_MS = 10000; /** * Blocks the threads awaiting until the agentQueues @@ -126,7 +127,6 @@ public class PubQueueCache { private Thread seeder; - public PubQueueCache(MessagingProvider messagingProvider, EventAdmin eventAdmin, DistributionMetricsService distributionMetricsService, String topic) { this.messagingProvider = messagingProvider; this.eventAdmin = eventAdmin; @@ -166,18 +166,23 @@ public class PubQueueCache { LOG.debug("Send seeding message"); try { sender.send(topic, pkgMsg); + sleep(SEEDING_DELAY_MS); } catch (MessagingException e) { LOG.warn(e.getMessage(), e); - } - try { - Thread.sleep(SEEDING_DELAY_MS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); + sleep(SEEDING_ERROR_DELAY_MS); } } LOG.info("Stop message seeder"); } + private void sleep(long sleepMs) { + try { + Thread.sleep(sleepMs); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + private PackageMessage createTestMessage() { String pkgId = UUID.randomUUID().toString(); return PackageMessage.newBuilder() diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/shared/JournalAvailableChecker.java b/src/main/java/org/apache/sling/distribution/journal/impl/shared/JournalAvailableChecker.java index c7397c0..1ac1e50 100644 --- a/src/main/java/org/apache/sling/distribution/journal/impl/shared/JournalAvailableChecker.java +++ b/src/main/java/org/apache/sling/distribution/journal/impl/shared/JournalAvailableChecker.java @@ -27,11 +27,9 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.io.IOUtils; import org.apache.sling.distribution.journal.ExceptionEventSender; -import org.apache.sling.distribution.journal.JournalAvailable; import org.apache.sling.distribution.journal.MessagingProvider; import org.apache.sling.distribution.journal.impl.shared.DistributionMetricsService.GaugeService; import org.osgi.framework.BundleContext; -import org.osgi.framework.ServiceRegistration; import org.osgi.service.component.annotations.Activate; import org.osgi.service.component.annotations.Component; import org.osgi.service.component.annotations.Deactivate; @@ -45,7 +43,7 @@ import org.slf4j.LoggerFactory; @Component( property = EventConstants.EVENT_TOPIC + "=" + ExceptionEventSender.ERROR_TOPIC ) -public class JournalAvailableChecker implements JournalAvailable, EventHandler { +public class JournalAvailableChecker implements EventHandler { private static final Duration INITIAL_RETRY_DELAY = Duration.of(1, SECONDS); private static final Duration MAX_RETRY_DELAY = Duration.of(5, MINUTES); @@ -68,9 +66,7 @@ public class JournalAvailableChecker implements JournalAvailable, EventHandler { @Reference DistributionMetricsService metrics; - private BundleContext context; - - private volatile ServiceRegistration<JournalAvailable> reg; + private JournalAvailableServiceMarker marker; private GaugeService<Boolean> gauge; @@ -83,7 +79,7 @@ public class JournalAvailableChecker implements JournalAvailable, EventHandler { public void activate(BundleContext context) { requireNonNull(provider); requireNonNull(topics); - this.context = context; + this.marker = new JournalAvailableServiceMarker(context); this.backoffRetry.startChecks(); this.gauge = metrics.createGauge(DistributionMetricsService.BASE_COMPONENT + ".journal_available", "", this::isAvailable); LOG.info("Started Journal availability checker service"); @@ -92,7 +88,7 @@ public class JournalAvailableChecker implements JournalAvailable, EventHandler { @Deactivate public void deactivate() { gauge.close(); - unRegister(); + this.marker.unRegister(); IOUtils.closeQuietly(this.backoffRetry); LOG.info("Stopped Journal availability checker service"); } @@ -106,11 +102,10 @@ public class JournalAvailableChecker implements JournalAvailable, EventHandler { private void available() { LOG.info("Journal is available"); - if (this.reg == null) { - this.reg = context.registerService(JournalAvailable.class, this, null); - } + this.numErrors.set(0); + this.marker.register(); } - + private void stillUnAvailable(Exception e) { String msg = "Journal is still unavailable: " + e.getMessage(); if (LOG.isDebugEnabled()) { @@ -118,11 +113,11 @@ public class JournalAvailableChecker implements JournalAvailable, EventHandler { } else { LOG.warn(msg); } - unRegister(); + this.marker.unRegister(); } public boolean isAvailable() { - return reg != null; + return this.marker.isRegistered(); } public void run() { @@ -136,22 +131,14 @@ public class JournalAvailableChecker implements JournalAvailable, EventHandler { } } - private void unRegister() { - if (this.reg != null) { - this.reg.unregister(); - this.reg = null; - } - } - @Override public synchronized void handleEvent(Event event) { String type = (String) event.getProperty(ExceptionEventSender.KEY_TYPE); int curNumErrors = this.numErrors.incrementAndGet(); - if (curNumErrors >= MIN_ERRORS) { + if (curNumErrors == MIN_ERRORS) { LOG.warn("Received exception event {}. Journal is considered unavailable.", type); - unRegister(); - this.numErrors.set(0); - this.backoffRetry.startChecks(); + this.marker.unRegister(); + this.backoffRetry.startChecks(); } else { LOG.info("Received exception event {}. {} of {} errors occurred.", type, curNumErrors, MIN_ERRORS); } diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/shared/JournalAvailableServiceMarker.java b/src/main/java/org/apache/sling/distribution/journal/impl/shared/JournalAvailableServiceMarker.java new file mode 100644 index 0000000..c6ac778 --- /dev/null +++ b/src/main/java/org/apache/sling/distribution/journal/impl/shared/JournalAvailableServiceMarker.java @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.sling.distribution.journal.impl.shared; + +import org.apache.sling.distribution.journal.JournalAvailable; +import org.osgi.framework.BundleContext; +import org.osgi.framework.ServiceRegistration; + +class JournalAvailableServiceMarker implements JournalAvailable { + + private BundleContext context; + private ServiceRegistration<JournalAvailable> reg; + + JournalAvailableServiceMarker(BundleContext context) { + this.context = context; + } + + synchronized void register() { + if (this.reg == null) { + this.reg = context.registerService(JournalAvailable.class, this, null); + } + } + + synchronized void unRegister() { + if (this.reg != null) { + this.reg.unregister(); + this.reg = null; + } + } + + synchronized boolean isRegistered() { + return this.reg != null; + } +}
