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 2023/10/31 14:47:06 UTC
(sling-org-apache-sling-distribution-journal) 01/01: SLING-12125 - Remove JournalAvailableCheck
This is an automated email from the ASF dual-hosted git repository.
cschneider pushed a commit to branch SLING-12125
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git
commit dc36091663503fd296438ef849cfb465b2720056
Author: Christian Schneider <cs...@adobe.com>
AuthorDate: Tue Oct 31 15:46:51 2023 +0100
SLING-12125 - Remove JournalAvailableCheck
---
.../journal/impl/discovery/DiscoveryService.java | 12 +-
.../impl/publisher/DistributionPublisher.java | 3 -
.../impl/publisher/PubQueueProviderPublisher.java | 3 -
.../impl/subscriber/DistributionSubscriber.java | 3 -
.../journal/shared/JournalAvailableChecker.java | 156 ----------------
.../shared/JournalAvailableServiceMarker.java | 53 ------
.../shared/JournalAvailableCheckerTest.java | 199 ---------------------
7 files changed, 4 insertions(+), 425 deletions(-)
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/discovery/DiscoveryService.java b/src/main/java/org/apache/sling/distribution/journal/impl/discovery/DiscoveryService.java
index 1ec0829..ef76bd1 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/discovery/DiscoveryService.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/discovery/DiscoveryService.java
@@ -30,6 +30,10 @@ import java.util.Hashtable;
import javax.annotation.ParametersAreNonnullByDefault;
+import org.apache.commons.io.IOUtils;
+import org.apache.sling.distribution.journal.MessageInfo;
+import org.apache.sling.distribution.journal.MessagingProvider;
+import org.apache.sling.distribution.journal.Reset;
import org.apache.sling.distribution.journal.messages.DiscoveryMessage;
import org.apache.sling.distribution.journal.messages.LogMessage;
import org.apache.sling.distribution.journal.messages.SubscriberConfig;
@@ -37,7 +41,6 @@ import org.apache.sling.distribution.journal.messages.SubscriberState;
import org.apache.sling.distribution.journal.shared.AgentId;
import org.apache.sling.distribution.journal.shared.PublisherConfigurationAvailable;
import org.apache.sling.distribution.journal.shared.Topics;
-import org.apache.commons.io.IOUtils;
import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceRegistration;
import org.osgi.service.component.annotations.Activate;
@@ -48,10 +51,6 @@ import org.osgi.service.component.annotations.ReferenceCardinality;
import org.osgi.service.component.annotations.ReferencePolicyOption;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
-import org.apache.sling.distribution.journal.MessageInfo;
-import org.apache.sling.distribution.journal.MessagingProvider;
-import org.apache.sling.distribution.journal.JournalAvailable;
-import org.apache.sling.distribution.journal.Reset;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -75,9 +74,6 @@ public class DiscoveryService implements Runnable {
private static final long REFRESH_TTL_MS = 30000;
- @Reference
- private JournalAvailable journalAvailable;
-
@Reference
private PublisherConfigurationAvailable publisherConfigurationAvailable;
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisher.java b/src/main/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisher.java
index ab49ea4..34f5851 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisher.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisher.java
@@ -111,9 +111,6 @@ public class DistributionPublisher implements DistributionAgent {
@Reference
private Topics topics;
- @Reference
- JournalAvailable journalAvailable;
-
@Reference
private DistributionMetricsService distributionMetricsService;
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/publisher/PubQueueProviderPublisher.java b/src/main/java/org/apache/sling/distribution/journal/impl/publisher/PubQueueProviderPublisher.java
index dd390d2..2e36cab 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/publisher/PubQueueProviderPublisher.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/publisher/PubQueueProviderPublisher.java
@@ -54,9 +54,6 @@ public class PubQueueProviderPublisher {
@Reference
private Topics topics;
- @Reference
- JournalAvailable journalAvailable;
-
@Reference
private DistributionMetricsService distributionMetricsService;
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 28c503e..5fadcff 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
@@ -117,9 +117,6 @@ public class DistributionSubscriber {
@Reference
private Topics topics;
- @Reference
- private JournalAvailable journalAvailable;
-
@Reference(name = "precondition")
private Precondition precondition;
diff --git a/src/main/java/org/apache/sling/distribution/journal/shared/JournalAvailableChecker.java b/src/main/java/org/apache/sling/distribution/journal/shared/JournalAvailableChecker.java
deleted file mode 100644
index 70f1529..0000000
--- a/src/main/java/org/apache/sling/distribution/journal/shared/JournalAvailableChecker.java
+++ /dev/null
@@ -1,156 +0,0 @@
-/*
- * 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.shared;
-
-import java.util.Arrays;
-
-import static java.util.Objects.requireNonNull;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.sling.distribution.journal.ExceptionEventSender;
-import org.apache.sling.distribution.journal.MessagingProvider;
-import org.osgi.framework.BundleContext;
-import org.osgi.service.component.annotations.Activate;
-import org.osgi.service.component.annotations.Component;
-import org.osgi.service.component.annotations.Deactivate;
-import org.osgi.service.component.annotations.Reference;
-import org.osgi.service.event.Event;
-import org.osgi.service.event.EventConstants;
-import org.osgi.service.event.EventHandler;
-import org.osgi.service.metatype.annotations.AttributeDefinition;
-import org.osgi.service.metatype.annotations.Designate;
-import org.osgi.service.metatype.annotations.ObjectClassDefinition;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Component(
- immediate = true,
- service = EventHandler.class,
- property = EventConstants.EVENT_TOPIC + "=" + ExceptionEventSender.ERROR_TOPIC
-)
-@Designate(ocd = JournalAvailableChecker.JournalCheckerConfiguration.class)
-public class JournalAvailableChecker implements EventHandler {
-
- public static final long INITIAL_RETRY_DELAY = 60000; // 1 minute
-
- public static final long MAX_RETRY_DELAY = 1800000; // 30 minutes
-
- private static final Logger LOG = LoggerFactory.getLogger(JournalAvailableChecker.class);
-
- private ExponentialBackOff backoffRetry;
-
- @Reference
- Topics topics;
-
- @Reference
- MessagingProvider provider;
-
- @Reference
- DistributionMetricsService metrics;
-
- JournalAvailableServiceMarker marker;
-
- @Activate
- public void activate(JournalCheckerConfiguration config, BundleContext context) {
- requireNonNull(provider);
- requireNonNull(topics);
- this.backoffRetry = new ExponentialBackOff(config.initialRetryDelay(), config.maxRetryDelay(), true, this::run);
- this.marker = new JournalAvailableServiceMarker(context);
- metrics.createGauge(DistributionMetricsService.BASE_COMPONENT + ".journal_available", this::isAvailable);
-
- Arrays.asList(config.trackedErrCodes()).stream().spliterator()
- .forEachRemaining(code -> metrics.getJournalErrorCodeCount(code));
-
- this.marker.register();
- LOG.info("Started Journal availability checker service with initialRetryDelay {}, maxRetryDelay {}. Journal is initially assumed available.", config.initialRetryDelay(), config.maxRetryDelay());
- }
-
- @Deactivate
- public void deactivate() {
- this.marker.unRegister();
- IOUtils.closeQuietly(this.backoffRetry);
- LOG.info("Stopped Journal availability checker service");
- }
-
- private void doChecks() {
- provider.assertTopic(topics.getPackageTopic());
- provider.assertTopic(topics.getDiscoveryTopic());
- provider.assertTopic(topics.getStatusTopic());
- provider.assertTopic(topics.getCommandTopic());
- }
-
- private void available() {
- LOG.info("Journal is available");
- this.marker.register();
- }
-
- private void stillUnAvailable(Exception e) {
- String msg = "Journal is still unavailable: " + e.getMessage();
- LOG.warn(msg, e);
- this.marker.unRegister();
- }
-
- public boolean isAvailable() {
- return this.marker.isRegistered();
- }
-
- public void run() {
- try {
- LOG.debug("Journal checker is running");
- doChecks();
- available();
- } catch (Exception e) {
- stillUnAvailable(e);
- throw e;
- }
- }
-
- @Override
- public synchronized void handleEvent(Event event) {
- String type = (String) event.getProperty(ExceptionEventSender.KEY_TYPE);
- String msg = (String) event.getProperty(ExceptionEventSender.KEY_MESSAGE);
- if (this.marker.isRegistered()) {
- LOG.warn("Received exception event {}: {}. Journal is considered unavailable.", type, msg);
- this.marker.unRegister();
- this.backoffRetry.startChecks();
- } else {
- LOG.info("Received exception event {}: {}. Journal still unavailable.", type, msg);
- }
- String errCode = (String) event.getProperty(ExceptionEventSender.KEY_ERROR_CODE);
- if ((errCode != null) && !errCode.isEmpty()) {
- metrics.getJournalErrorCodeCount(errCode).increment();
- }
- }
-
- @ObjectClassDefinition(name = "Apache Sling Journal based Distribution - Journal Checker")
- public @interface JournalCheckerConfiguration {
-
- @AttributeDefinition(name = "Initial retry delay",
- description = "The initial retry delay in milliseconds.")
- long initialRetryDelay() default INITIAL_RETRY_DELAY;
-
- @AttributeDefinition(name = "Max retry delay",
- description = "The max retry delay in milliseconds.")
- long maxRetryDelay() default MAX_RETRY_DELAY;
-
- @AttributeDefinition(name ="Tracked response codes",
- description = "Response error codes tracked in metrics.")
- String[] trackedErrCodes() default {"400", "401", "404", "405", "413", "500", "503", "505"};
- }
-}
diff --git a/src/main/java/org/apache/sling/distribution/journal/shared/JournalAvailableServiceMarker.java b/src/main/java/org/apache/sling/distribution/journal/shared/JournalAvailableServiceMarker.java
deleted file mode 100644
index d18c1aa..0000000
--- a/src/main/java/org/apache/sling/distribution/journal/shared/JournalAvailableServiceMarker.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.shared;
-
-import java.util.Objects;
-
-import org.apache.sling.distribution.journal.JournalAvailable;
-import org.osgi.framework.BundleContext;
-import org.osgi.framework.ServiceRegistration;
-
-class JournalAvailableServiceMarker implements JournalAvailable {
-
- private final 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);
- Objects.requireNonNull(this.reg); // To make incomplete mocking visible in tests
- }
- }
-
- synchronized void unRegister() {
- if (this.reg != null) {
- this.reg.unregister();
- this.reg = null;
- }
- }
-
- synchronized boolean isRegistered() {
- return this.reg != null;
- }
-}
diff --git a/src/test/java/org/apache/sling/distribution/journal/shared/JournalAvailableCheckerTest.java b/src/test/java/org/apache/sling/distribution/journal/shared/JournalAvailableCheckerTest.java
deleted file mode 100644
index d699858..0000000
--- a/src/test/java/org/apache/sling/distribution/journal/shared/JournalAvailableCheckerTest.java
+++ /dev/null
@@ -1,199 +0,0 @@
-/*
- * 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.shared;
-
-import static java.util.Collections.emptyMap;
-import static java.util.Collections.singletonMap;
-import static org.awaitility.Awaitility.await;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.when;
-import static org.osgi.util.converter.Converters.standardConverter;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.sling.commons.metrics.Counter;
-import org.apache.sling.distribution.journal.ExceptionEventSender;
-import org.apache.sling.distribution.journal.JournalAvailable;
-import org.apache.sling.distribution.journal.MessagingException;
-import org.apache.sling.distribution.journal.MessagingProvider;
-import org.apache.sling.distribution.journal.shared.JournalAvailableChecker.JournalCheckerConfiguration;
-import org.apache.sling.distribution.journal.shared.Topics.TopicsConfiguration;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.Mockito;
-import org.mockito.Spy;
-import org.mockito.runners.MockitoJUnitRunner;
-import org.osgi.framework.BundleContext;
-import org.osgi.framework.ServiceRegistration;
-import org.osgi.service.event.Event;
-
-@RunWith(MockitoJUnitRunner.class)
-public class JournalAvailableCheckerTest {
-
- private static final String INVALID_TOPIC = "invalid_topic_name";
-
- @InjectMocks
- private JournalAvailableChecker checker;
-
- @Spy
- private Topics topics = new Topics();
-
- @Mock
- private MessagingProvider provider;
-
- @Mock
- private DistributionMetricsService metrics;
-
- @Mock
- private BundleContext context;
-
- @Mock
- private ServiceRegistration<JournalAvailable> sreg;
-
- @Before
- public void before() throws Exception {
- MessagingException me = new MessagingException("expected", new RuntimeException("expected nested exception"));
- doThrow(me)
- .when(provider).assertTopic(INVALID_TOPIC);
- when(context.registerService(Mockito.eq(JournalAvailable.class), Mockito.any(JournalAvailable.class), Mockito.any()))
- .thenReturn(sreg);
- Map<String, String> config = new HashMap<>();
- config.put("initialRetryDelay", "500");
- config.put("maxRetryDelay", "300000");
- checker.activate(configuration(config, JournalCheckerConfiguration.class), context);
- }
-
- @After
- public void after() throws Exception {
- checker.deactivate();
- }
-
- @Test
- public void testIsAvailable() throws Exception {
- makeCheckFail();
- try {
- checker.run();
- Assert.fail("Should throw exception");
- } catch (Exception e) {
- }
- assertFalse(checker.isAvailable());
- makeCheckSucceed();
- checker.run();
- assertTrue(checker.isAvailable());
- }
-
- @Test
- public void testActivateChecksOnEvent() throws InterruptedException {
- activateChecksOnEvent(new MessagingException("Expected", "400"), 1);
- }
-
- @Test
- public void testActivateChecksOnEventNoErrCode() throws InterruptedException {
- activateChecksOnEvent(new MessagingException("Expected"), 0);
- }
-
- @Test
- public void testActivateChecksOnEventOtherException() throws InterruptedException {
- activateChecksOnEvent(new IOException("Expected"), 0);
- }
-
- @Test
- public void testActivateChecksOnEventWithoutMessage() throws InterruptedException {
- activateChecksOnEvent(new IOException(), 0);
- }
-
- public void activateChecksOnEvent(Exception e, int expectedCounter) throws InterruptedException {
- Counter counter = new TestCounter();
- when(metrics.getJournalErrorCodeCount("400")).thenReturn(counter);
-
- await("At the start checks are triggers and should set the state available")
- .until(checker::isAvailable);
-
- makeCheckFail();
- Event event = createErrorEvent(e);
- checker.handleEvent(event);
- assertEquals(expectedCounter, counter.getCount());
- await().until(() -> !checker.isAvailable());
- Thread.sleep(1000); // Make sure we get at least one failed doCheck
- makeCheckSucceed();
- await().until(checker::isAvailable);
- }
-
- private void makeCheckSucceed() {
- topics.activate(configuration(emptyMap(), TopicsConfiguration.class));
- }
-
- private void makeCheckFail() {
- topics.activate(configuration(singletonMap("packageTopic", INVALID_TOPIC), TopicsConfiguration.class));
- }
-
- private <T> T configuration(Map<String,String> props, Class<T> clazz) {
- return standardConverter()
- .convert(props)
- .to(clazz);
- }
-
- private static Event createErrorEvent(Exception e) {
- Map<String, String> props = new HashMap<>();
- props.put(ExceptionEventSender.KEY_TYPE, e.getClass().getName());
- props.put(ExceptionEventSender.KEY_MESSAGE, e.getMessage());
- if (e instanceof MessagingException) {
- props.put(ExceptionEventSender.KEY_ERROR_CODE, ((MessagingException) e).getResponseCode());
- }
- return new Event(ExceptionEventSender.ERROR_TOPIC, props);
- }
-
- class TestCounter implements Counter {
- AtomicLong l = new AtomicLong();
- @Override public void increment() {
- l.getAndIncrement();
- }
-
- @Override public void decrement() {
- l.decrementAndGet();
- }
-
- @Override public void increment(long n) {
- l.addAndGet(n);
- }
-
- @Override public void decrement(long n) {
- l.addAndGet(-n);
- }
-
- @Override public long getCount() {
- return l.get();
- }
-
- @Override public <A> A adaptTo(Class<A> type) {
- return null;
- }
- }
-}