You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sling.apache.org by st...@apache.org on 2021/09/14 14:40:55 UTC
[sling-org-apache-sling-discovery-commons] branch master updated:
SLING-10489 : ignore partially started instances : (#4)
This is an automated email from the ASF dual-hosted git repository.
stefanegli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-discovery-commons.git
The following commit(s) were added to refs/heads/master by this push:
new d62c7da SLING-10489 : ignore partially started instances : (#4)
d62c7da is described below
commit d62c7da07909373098c6ad4c2bc54124b0b4017f
Author: stefan-egli <st...@apache.org>
AuthorDate: Tue Sep 14 16:40:38 2021 +0200
SLING-10489 : ignore partially started instances : (#4)
* SLING-10489 : ignore partially started instances : a. skipping activeIds in OakBacklogClusterSyncService if they are partially started, b. ignore syncToken for view change checks if there are partially started instances plus bonus c. LogSilencer introduced, which reduces log.info spam caused by discovery
* SLING-10489 : removed a noisy log.info
* SLING-10489 : lowered a noisy log.info
* SLING-10489 : lowered a noisy log.info
* SLING-10489 : lowered a noisy log.info
* SLING-10489: Ignore partially started, newly joining instances to avoid disturbing discovery (for a while)
Unit test for LocalClusterView
* Update src/main/java/org/apache/sling/discovery/commons/providers/util/LogSilencer.java
Co-authored-by: Marcel Reutegger <ma...@gmail.com>
* Update src/main/java/org/apache/sling/discovery/commons/providers/util/LogSilencer.java
Co-authored-by: Marcel Reutegger <ma...@gmail.com>
* SLING-10489 : javadoc updated
* SLING-10489 : copy partiallyStartedClusterNodeIds on clone as well
* SLING-10489 : simplified equalsIgnoreSyncToken - plus a unit test added for it
Co-authored-by: Marcel Reutegger <mr...@adobe.com>
Co-authored-by: Marcel Reutegger <ma...@gmail.com>
---
.../providers/base/MinEventDelayHandler.java | 37 +++--
.../providers/base/ViewStateManagerImpl.java | 45 +++++-
.../commons/providers/base/package-info.java | 4 +-
.../commons/providers/spi/LocalClusterView.java | 27 ++++
.../base/AbstractServiceWithBackgroundCheck.java | 16 +-
.../spi/base/OakBacklogClusterSyncService.java | 40 +++--
.../providers/spi/base/SyncTokenService.java | 29 ++--
.../commons/providers/spi/package-info.java | 4 +-
.../commons/providers/util/LogSilencer.java | 97 ++++++++++++
.../commons/providers/util/package-info.java | 4 +-
.../commons/providers/DummyTopologyView.java | 11 +-
.../providers/base/TestViewStateManager.java | 173 +++++++++++++++++++++
.../providers/spi/LocalClusterViewTest.java | 46 ++++++
.../spi/base/TestOakSyncTokenService.java | 84 +++++++++-
14 files changed, 566 insertions(+), 51 deletions(-)
diff --git a/src/main/java/org/apache/sling/discovery/commons/providers/base/MinEventDelayHandler.java b/src/main/java/org/apache/sling/discovery/commons/providers/base/MinEventDelayHandler.java
index 43093ad..12a4dfe 100644
--- a/src/main/java/org/apache/sling/discovery/commons/providers/base/MinEventDelayHandler.java
+++ b/src/main/java/org/apache/sling/discovery/commons/providers/base/MinEventDelayHandler.java
@@ -25,6 +25,7 @@ import org.apache.sling.commons.scheduler.Scheduler;
import org.apache.sling.discovery.DiscoveryService;
import org.apache.sling.discovery.TopologyView;
import org.apache.sling.discovery.commons.providers.BaseTopologyView;
+import org.apache.sling.discovery.commons.providers.util.LogSilencer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,6 +52,8 @@ class MinEventDelayHandler {
private volatile int cancelCnt = 0;
+ private final LogSilencer logSilencer = new LogSilencer(logger);
+
MinEventDelayHandler(ViewStateManagerImpl viewStateManager, Lock lock,
DiscoveryService discoveryService, Scheduler scheduler,
long minEventDelaySecs) {
@@ -74,23 +77,33 @@ class MinEventDelayHandler {
* Asks the MinEventDelayHandler to handle the new view
* and return true if the caller shouldn't worry about any follow-up action -
* only if the method returns false should the caller do the usual
- * handleNewView action
+ * handleNewView action.
+ * This caller of this method must ensure to be in a lock.lock() block
*/
boolean handlesNewView(BaseTopologyView newView) {
if (isDelaying) {
// already delaying, so we'll soon ask the DiscoveryServiceImpl for the
// latest view and go ahead then
- logger.info("handleNewView: already delaying, ignoring new view meanwhile");
+ logSilencer.infoOrDebug("handlesNewView-" + newView.getLocalClusterSyncTokenId(),
+ "handleNewView: already delaying, ignoring new view meanwhile");
return true;
}
if (!viewStateManager.hadPreviousView()) {
- logger.info("handlesNewView: never had a previous view, hence no delaying applicable");
+ logSilencer.infoOrDebug("handlesNewView-" + newView.getLocalClusterSyncTokenId(),
+ "handlesNewView: never had a previous view, hence no delaying applicable");
+ return false;
+ }
+
+ if (viewStateManager.equalsIgnoreSyncToken(newView)) {
+ // this is a frequent case, hence only log.debug
+ logger.debug("handlesNewView: equalsIgnoreSyncToken, hence no delaying applicable");
return false;
}
if (viewStateManager.onlyDiffersInProperties(newView)) {
- logger.info("handlesNewView: only properties differ, hence no delaying applicable");
+ logSilencer.infoOrDebug("handlesNewView-" + newView.getLocalClusterSyncTokenId(),
+ "handlesNewView: only properties differ, hence no delaying applicable");
return false;
}
@@ -103,7 +116,8 @@ class MinEventDelayHandler {
// thanks to force==true this will always return true
if (!triggerAsyncDelaying(newView)) {
- logger.info("handleNewView: could not trigger async delaying, sending new view now.");
+ logSilencer.infoOrDebug("handlesNewView-" + newView.getLocalClusterSyncTokenId(),
+ "handleNewView: could not trigger async delaying, sending new view now.");
viewStateManager.handleNewViewNonDelayed(newView);
} else {
// if triggering the async event was successful, then we should also
@@ -118,7 +132,7 @@ class MinEventDelayHandler {
return true;
}
- private boolean triggerAsyncDelaying(BaseTopologyView newView) {
+ private boolean triggerAsyncDelaying(final BaseTopologyView newView) {
final int validCancelCnt = cancelCnt;
final boolean triggered = runAfter(minEventDelaySecs /*seconds*/ , new Runnable() {
@@ -127,7 +141,8 @@ class MinEventDelayHandler {
lock.lock();
try{
if (cancelCnt!=validCancelCnt) {
- logger.info("asyncDelay.run: got cancelled (validCancelCnt="+validCancelCnt+", cancelCnt="+cancelCnt+"), quitting.");
+ logSilencer.infoOrDebug("asyncDelay.run-cancel-" + newView.getLocalClusterSyncTokenId(),
+ "asyncDelay.run: got cancelled (validCancelCnt="+validCancelCnt+", cancelCnt="+cancelCnt+"), quitting.");
return;
}
@@ -144,10 +159,12 @@ class MinEventDelayHandler {
BaseTopologyView topology = (BaseTopologyView) t;
if (topology.isCurrent()) {
- logger.info("asyncDelay.run: done delaying. got new view: "+ topology.toShortString());
+ logSilencer.infoOrDebug("asyncDelay.run-done-" + newView.getLocalClusterSyncTokenId(),
+ "asyncDelay.run: done delaying. got new view: "+ topology.toShortString());
viewStateManager.handleNewViewNonDelayed(topology);
} else {
- logger.info("asyncDelay.run: done delaying. new view (still/again) not current, delaying again");
+ logSilencer.infoOrDebug("asyncDelay.run-done-" + newView.getLocalClusterSyncTokenId(),
+ "asyncDelay.run: done delaying. new view (still/again) not current, delaying again");
triggerAsyncDelaying(topology);
// we're actually not interested in the result here
// if the async part failed, then we have to rely
@@ -168,7 +185,7 @@ class MinEventDelayHandler {
}
});
- logger.info("triggerAsyncDelaying: asynch delaying of "+minEventDelaySecs+" triggered: "+triggered);
+ logSilencer.infoOrDebug("triggerAsyncDelaying", "triggerAsyncDelaying: asynch delaying of "+minEventDelaySecs+" triggered: "+triggered);
if (triggered) {
isDelaying = true;
}
diff --git a/src/main/java/org/apache/sling/discovery/commons/providers/base/ViewStateManagerImpl.java b/src/main/java/org/apache/sling/discovery/commons/providers/base/ViewStateManagerImpl.java
index b9f9fc7..da47195 100644
--- a/src/main/java/org/apache/sling/discovery/commons/providers/base/ViewStateManagerImpl.java
+++ b/src/main/java/org/apache/sling/discovery/commons/providers/base/ViewStateManagerImpl.java
@@ -28,6 +28,7 @@ import java.util.Set;
import java.util.concurrent.locks.Lock;
import org.apache.sling.commons.scheduler.Scheduler;
+import org.apache.sling.discovery.ClusterView;
import org.apache.sling.discovery.DiscoveryService;
import org.apache.sling.discovery.InstanceDescription;
import org.apache.sling.discovery.TopologyEvent;
@@ -37,6 +38,8 @@ import org.apache.sling.discovery.commons.providers.BaseTopologyView;
import org.apache.sling.discovery.commons.providers.EventHelper;
import org.apache.sling.discovery.commons.providers.ViewStateManager;
import org.apache.sling.discovery.commons.providers.spi.ClusterSyncService;
+import org.apache.sling.discovery.commons.providers.spi.LocalClusterView;
+import org.apache.sling.discovery.commons.providers.util.LogSilencer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -137,6 +140,8 @@ public class ViewStateManagerImpl implements ViewStateManager {
private MinEventDelayHandler minEventDelayHandler;
+ private final LogSilencer logSilencer = new LogSilencer(logger);
+
/**
* Creates a new ViewStateManager which synchronizes each method with the given
* lock and which optionally uses the given ClusterSyncService to sync the repository
@@ -456,7 +461,7 @@ public class ViewStateManagerImpl implements ViewStateManager {
// verify if there is actually a change between previousView and newView
// if there isn't, then there is not much point in sending a CHANGING/CHANGED tuple
// at all
- if (previousView!=null && previousView.equals(newView)) {
+ if (previousView!=null && (previousView.equals(newView) || equalsIgnoreSyncToken(newView))) {
// then nothing to send - the view has not changed, and we haven't
// sent the CHANGING event - so we should not do anything here
logger.debug("handleNewViewNonDelayed: we were not in changing state and new view matches old, so - ignoring");
@@ -486,7 +491,7 @@ public class ViewStateManagerImpl implements ViewStateManager {
if (!isChanging && onlyDiffersInProperties(newView)) {
// well then send a properties changed event only
// and that one does not go via consistencyservice
- logger.info("handleNewViewNonDelayed: properties changed to: "+newView);
+ logSilencer.infoOrDebug("handleNewViewNonDelayed-propsChanged", "handleNewViewNonDelayed: properties changed to: "+newView);
previousView.setNotCurrent();
enqueueForAll(eventListeners, EventHelper.newPropertiesChangedEvent(previousView, newView));
logger.trace("handleNewViewNonDelayed: setting previousView to {}", newView);
@@ -496,7 +501,7 @@ public class ViewStateManagerImpl implements ViewStateManager {
final boolean invokeClusterSyncService;
if (consistencyService==null) {
- logger.info("handleNewViewNonDelayed: no ClusterSyncService set - continuing directly.");
+ logSilencer.infoOrDebug("handleNewViewNonDelayed-noSyncService", "handleNewViewNonDelayed: no ClusterSyncService set - continuing directly.");
invokeClusterSyncService = false;
} else {
// there used to be a distinction between:
@@ -511,7 +516,7 @@ public class ViewStateManagerImpl implements ViewStateManager {
//
// which is a long way of saying: if the consistencyService is configured,
// then we always use it, hence:
- logger.info("handleNewViewNonDelayed: ClusterSyncService set - invoking...");
+ logSilencer.infoOrDebug("handleNewViewNonDelayed-invokeSyncService", "handleNewViewNonDelayed: ClusterSyncService set - invoking...");
invokeClusterSyncService = true;
}
@@ -520,7 +525,7 @@ public class ViewStateManagerImpl implements ViewStateManager {
// then:
// run the set consistencyService
final int lastModCnt = modCnt;
- logger.info("handleNewViewNonDelayed: invoking waitForAsyncEvents, then clusterSyncService (modCnt={})", modCnt);
+ logSilencer.infoOrDebug("handleNewViewNonDelayed-invokeWaitAsync", "handleNewViewNonDelayed: invoking waitForAsyncEvents, then clusterSyncService");
asyncEventSender.enqueue(new AsyncEvent() {
@Override
@@ -544,7 +549,7 @@ public class ViewStateManagerImpl implements ViewStateManager {
lastModCnt, modCnt);
return;
}
- logger.info("handleNewViewNonDelayed/waitForAsyncEvents.run: done, now invoking consistencyService (modCnt={})", modCnt);
+ logSilencer.infoOrDebug("waitForAsyncEvents-asyncRun", "handleNewViewNonDelayed/waitForAsyncEvents.run: done, now invoking consistencyService");
consistencyService.sync(newView,
new Runnable() {
@@ -558,7 +563,7 @@ public class ViewStateManagerImpl implements ViewStateManager {
lastModCnt, modCnt);
return;
}
- logger.info("consistencyService.callback.run: invoking doHandleConsistent.");
+ logSilencer.infoOrDebug("consistencyService-callBackRun", "consistencyService.callback.run: invoking doHandleConsistent.");
// else:
doHandleConsistent(newView);
} finally {
@@ -579,7 +584,7 @@ public class ViewStateManagerImpl implements ViewStateManager {
// or using it is not applicable at this stage - so continue
// with sending the TOPOLOGY_CHANGED (or TOPOLOGY_INIT if there
// are any newly bound topology listeners) directly
- logger.info("handleNewViewNonDelayed: not invoking consistencyService, considering consistent now");
+ logSilencer.infoOrDebug("handleNewViewNonDelayed-noSyncService-ignore", "handleNewViewNonDelayed: not invoking consistencyService, considering consistent now");
doHandleConsistent(newView);
}
logger.debug("handleNewViewNonDelayed: end");
@@ -590,6 +595,30 @@ public class ViewStateManagerImpl implements ViewStateManager {
}
}
+ /**
+ * Checks if the previouesView is equal to the newView, ignoring the
+ * syncToken (but only if the newView has partially started instances).
+ * <p/>
+ * This caller of this method must ensure to be in a lock.lock() block
+ */
+ protected boolean equalsIgnoreSyncToken(BaseTopologyView newView) {
+ if (previousView==null) {
+ return false;
+ }
+ if (newView==null) {
+ throw new IllegalArgumentException("newView must not be null");
+ }
+ final ClusterView cluster = newView.getLocalInstance().getClusterView();
+ if (cluster instanceof LocalClusterView) {
+ final LocalClusterView local = (LocalClusterView)cluster;
+ if (!local.hasPartiallyStartedInstances()) {
+ // then we should not ignore the syncToken I'm afraid
+ return previousView.equals(newView);
+ }
+ }
+ return previousView.getInstances().equals(newView.getInstances());
+ }
+
protected boolean onlyDiffersInProperties(BaseTopologyView newView) {
if (previousView==null) {
return false;
diff --git a/src/main/java/org/apache/sling/discovery/commons/providers/base/package-info.java b/src/main/java/org/apache/sling/discovery/commons/providers/base/package-info.java
index 9d9bd43..fd0915d 100644
--- a/src/main/java/org/apache/sling/discovery/commons/providers/base/package-info.java
+++ b/src/main/java/org/apache/sling/discovery/commons/providers/base/package-info.java
@@ -20,9 +20,9 @@
/**
* Provides commons implementations for providers of the Discovery API.
*
- * @version 1.0.0
+ * @version 1.1.0
*/
-@Version("1.0.0")
+@Version("1.1.0")
package org.apache.sling.discovery.commons.providers.base;
import org.osgi.annotation.versioning.Version;
diff --git a/src/main/java/org/apache/sling/discovery/commons/providers/spi/LocalClusterView.java b/src/main/java/org/apache/sling/discovery/commons/providers/spi/LocalClusterView.java
index 12363af..29663c0 100644
--- a/src/main/java/org/apache/sling/discovery/commons/providers/spi/LocalClusterView.java
+++ b/src/main/java/org/apache/sling/discovery/commons/providers/spi/LocalClusterView.java
@@ -18,11 +18,17 @@
*/
package org.apache.sling.discovery.commons.providers.spi;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
import org.apache.sling.discovery.commons.providers.DefaultClusterView;
public class LocalClusterView extends DefaultClusterView {
private final String localClusterSyncTokenId;
+ private Set<Integer> partiallyStartedClusterNodeIds;
public LocalClusterView(String id, String localClusterSyncTokenId) {
super(id);
@@ -33,4 +39,25 @@ public class LocalClusterView extends DefaultClusterView {
return localClusterSyncTokenId;
}
+ public Set<Integer> getPartiallyStartedClusterNodeIds() {
+ return Collections.unmodifiableSet(partiallyStartedClusterNodeIds);
+ }
+
+ public void setPartiallyStartedClusterNodeIds(Collection<Integer> clusterNodeIds) {
+ this.partiallyStartedClusterNodeIds = new HashSet<Integer>(clusterNodeIds);
+ }
+
+ public boolean isPartiallyStarted(Integer clusterNodeId) {
+ if (partiallyStartedClusterNodeIds == null || clusterNodeId == null) {
+ return false;
+ }
+ return partiallyStartedClusterNodeIds.contains(clusterNodeId);
+ }
+
+ public boolean hasPartiallyStartedInstances() {
+ if (partiallyStartedClusterNodeIds == null) {
+ return false;
+ }
+ return !partiallyStartedClusterNodeIds.isEmpty();
+ }
}
diff --git a/src/main/java/org/apache/sling/discovery/commons/providers/spi/base/AbstractServiceWithBackgroundCheck.java b/src/main/java/org/apache/sling/discovery/commons/providers/spi/base/AbstractServiceWithBackgroundCheck.java
index 84f2f76..e7a2e5c 100644
--- a/src/main/java/org/apache/sling/discovery/commons/providers/spi/base/AbstractServiceWithBackgroundCheck.java
+++ b/src/main/java/org/apache/sling/discovery/commons/providers/spi/base/AbstractServiceWithBackgroundCheck.java
@@ -18,6 +18,7 @@
*/
package org.apache.sling.discovery.commons.providers.spi.base;
+import org.apache.sling.discovery.commons.providers.util.LogSilencer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -32,6 +33,8 @@ public abstract class AbstractServiceWithBackgroundCheck {
protected String slingId;
+ private final LogSilencer logSilencer = new LogSilencer(logger);
+
/**
* The BackgroundCheckRunnable implements the details of
* calling BackgroundCheck.check and looping until it
@@ -83,9 +86,9 @@ public abstract class AbstractServiceWithBackgroundCheck {
if (timeoutMillis != -1 &&
(System.currentTimeMillis() > start + timeoutMillis)) {
if (callback == null) {
- logger.info("backgroundCheck.run: timeout hit (no callback to invoke)");
+ logSilencer.infoOrDebug("backgroundCheck.run", "backgroundCheck.run: timeout hit (no callback to invoke)");
} else {
- logger.info("backgroundCheck.run: timeout hit, invoking callback.");
+ logSilencer.infoOrDebug("backgroundCheck.run", "backgroundCheck.run: timeout hit, invoking callback.");
callback.run();
}
return;
@@ -131,7 +134,7 @@ public abstract class AbstractServiceWithBackgroundCheck {
void cancel() {
if (!done) {
- logger.info("cancel: "+threadName);
+ logSilencer.infoOrDebug("cancel-" + threadName, "cancel: "+threadName);
}
cancelled = true;
}
@@ -200,14 +203,15 @@ public abstract class AbstractServiceWithBackgroundCheck {
// then we're not even going to start the background-thread
// we're already done
if (callback!=null) {
- logger.info("backgroundCheck: already done, backgroundCheck successful, invoking callback");
+ logSilencer.infoOrDebug("backgroundCheck", "backgroundCheck: already done, backgroundCheck successful, invoking callback");
callback.run();
} else {
- logger.info("backgroundCheck: already done, backgroundCheck successful. no callback to invoke.");
+ logSilencer.infoOrDebug("backgroundCheck", "backgroundCheck: already done, backgroundCheck successful. no callback to invoke.");
}
return;
}
- logger.info("backgroundCheck: spawning background-thread for '"+threadName+"'");
+ logSilencer.infoOrDebug("backgroundCheck-" + threadName,
+ "backgroundCheck: spawning background-thread for '"+threadName+"'");
backgroundCheckRunnable = new BackgroundCheckRunnable(callback, check, timeoutMillis, waitMillis, threadName);
Thread th = new Thread(backgroundCheckRunnable);
th.setName(threadName);
diff --git a/src/main/java/org/apache/sling/discovery/commons/providers/spi/base/OakBacklogClusterSyncService.java b/src/main/java/org/apache/sling/discovery/commons/providers/spi/base/OakBacklogClusterSyncService.java
index f171728..abd94fb 100644
--- a/src/main/java/org/apache/sling/discovery/commons/providers/spi/base/OakBacklogClusterSyncService.java
+++ b/src/main/java/org/apache/sling/discovery/commons/providers/spi/base/OakBacklogClusterSyncService.java
@@ -30,6 +30,8 @@ import org.apache.sling.discovery.ClusterView;
import org.apache.sling.discovery.InstanceDescription;
import org.apache.sling.discovery.commons.providers.BaseTopologyView;
import org.apache.sling.discovery.commons.providers.spi.ClusterSyncService;
+import org.apache.sling.discovery.commons.providers.spi.LocalClusterView;
+import org.apache.sling.discovery.commons.providers.util.LogSilencer;
import org.apache.sling.settings.SlingSettingsService;
import org.osgi.framework.Constants;
import org.osgi.service.component.annotations.Activate;
@@ -67,6 +69,8 @@ public class OakBacklogClusterSyncService extends AbstractServiceWithBackgroundC
private ClusterSyncHistory consistencyHistory = new ClusterSyncHistory();
+ private final LogSilencer logSilencer = new LogSilencer(logger);
+
public static OakBacklogClusterSyncService testConstructorAndActivate(
final DiscoveryLiteConfig commonsConfig,
final IdMapService idMapService,
@@ -132,20 +136,21 @@ public class OakBacklogClusterSyncService extends AbstractServiceWithBackgroundC
cancelPreviousBackgroundCheck();
// first do the wait-for-backlog part
- logger.info("sync: doing wait-for-backlog part for view="+view.toShortString());
+ logSilencer.infoOrDebug("sync", "sync: doing wait-for-backlog part for view="+view.toShortString());
waitWhileBacklog(view, callback);
}
private void waitWhileBacklog(final BaseTopologyView view, final Runnable runnable) {
// start backgroundChecking until the backlogStatus
// is NO_BACKLOG
- startBackgroundCheck("OakBacklogClusterSyncService-backlog-waiting", new BackgroundCheck() {
+ startBackgroundCheck("OakBacklogClusterSyncService-backlog-waiting-" + view.getLocalClusterSyncTokenId(), new BackgroundCheck() {
@Override
public boolean check() {
try {
if (!idMapService.isInitialized()) {
- logger.info("waitWhileBacklog: could not initialize...");
+ logSilencer.infoOrDebug("waitWhileBacklog-" + view.toShortString(),
+ "waitWhileBacklog: could not initialize...");
consistencyHistory.addHistoryEntry(view, "could not initialize idMapService");
return false;
}
@@ -156,11 +161,13 @@ public class OakBacklogClusterSyncService extends AbstractServiceWithBackgroundC
}
BacklogStatus backlogStatus = getBacklogStatus(view);
if (backlogStatus == BacklogStatus.NO_BACKLOG) {
- logger.info("waitWhileBacklog: no backlog (anymore), done.");
+ logSilencer.infoOrDebug("waitWhileBacklog-" + view.toShortString(),
+ "waitWhileBacklog: no backlog (anymore), done.");
consistencyHistory.addHistoryEntry(view, "no backlog (anymore)");
return true;
} else {
- logger.info("waitWhileBacklog: backlogStatus still "+backlogStatus);
+ logSilencer.infoOrDebug("waitWhileBacklog-" + view.toShortString(),
+ "waitWhileBacklog: backlogStatus still "+backlogStatus);
// clear the cache to make sure to get the latest version in case something changed
idMapService.clearCache();
consistencyHistory.addHistoryEntry(view, "backlog status "+backlogStatus);
@@ -201,11 +208,16 @@ public class OakBacklogClusterSyncService extends AbstractServiceWithBackgroundC
// 1) 'deactivating' must be empty
if (deactivatingIds.length!=0) {
- logger.info("getBacklogStatus: there are deactivating instances: "+Arrays.toString(deactivatingIds));
+ logSilencer.infoOrDebug("getBacklogStatus-hasBacklog-" + view.toShortString(),
+ "getBacklogStatus: there are deactivating instances: "+Arrays.toString(deactivatingIds));
return BacklogStatus.HAS_BACKLOG;
}
ClusterView cluster = view.getLocalInstance().getClusterView();
+ LocalClusterView localCluster = null;
+ if (cluster instanceof LocalClusterView) {
+ localCluster = (LocalClusterView)cluster;
+ }
Set<String> slingIds = new HashSet<>();
for (InstanceDescription instance : cluster.getInstances()) {
slingIds.add(instance.getSlingId());
@@ -213,23 +225,31 @@ public class OakBacklogClusterSyncService extends AbstractServiceWithBackgroundC
for(int i=0; i<activeIds.length; i++) {
int activeId = activeIds[i];
+ if (localCluster != null && localCluster.isPartiallyStarted(activeId)) {
+ // ignore this one then
+ continue;
+ }
String slingId = idMapService.toSlingId(activeId, resourceResolver);
// 2) all ids of the descriptor must have a mapping to slingIds
if (slingId == null) {
- logger.info("getBacklogStatus: no slingId found for active id: "+activeId);
+ logSilencer.infoOrDebug("getBacklogStatus-undefined-" + view.toShortString(),
+ "getBacklogStatus: no slingId found for active id: "+activeId);
return BacklogStatus.UNDEFINED;
}
// 3) all 'active' instances must be in the view
if (!slingIds.contains(slingId)) {
- logger.info("getBacklogStatus: active instance's ("+activeId+") slingId ("+slingId+") not found in cluster ("+cluster+")");
+ logSilencer.infoOrDebug("getBacklogStatus-hasBacklog-" + view.toShortString(),
+ "getBacklogStatus: active instance's ("+activeId+") slingId ("+slingId+") not found in cluster ("+cluster+")");
return BacklogStatus.HAS_BACKLOG;
}
}
- logger.info("getBacklogStatus: no backlog (anymore)");
+ logSilencer.infoOrDebug("getBacklogStatus-" + view.toShortString(),
+ "getBacklogStatus: no backlog (anymore)");
return BacklogStatus.NO_BACKLOG;
} catch(Exception e) {
- logger.info("getBacklogStatus: failed to determine backlog status: "+e);
+ logSilencer.infoOrDebug("getBacklogStatus-undefined-" + view.toShortString(),
+ "getBacklogStatus: failed to determine backlog status: "+e);
return BacklogStatus.UNDEFINED;
} finally {
logger.trace("getBacklogStatus: end");
diff --git a/src/main/java/org/apache/sling/discovery/commons/providers/spi/base/SyncTokenService.java b/src/main/java/org/apache/sling/discovery/commons/providers/spi/base/SyncTokenService.java
index c5bc4d9..0fe5209 100644
--- a/src/main/java/org/apache/sling/discovery/commons/providers/spi/base/SyncTokenService.java
+++ b/src/main/java/org/apache/sling/discovery/commons/providers/spi/base/SyncTokenService.java
@@ -28,6 +28,7 @@ import org.apache.sling.api.resource.ValueMap;
import org.apache.sling.discovery.InstanceDescription;
import org.apache.sling.discovery.commons.providers.BaseTopologyView;
import org.apache.sling.discovery.commons.providers.spi.ClusterSyncService;
+import org.apache.sling.discovery.commons.providers.util.LogSilencer;
import org.apache.sling.discovery.commons.providers.util.ResourceHelper;
import org.apache.sling.settings.SlingSettingsService;
import org.osgi.framework.Constants;
@@ -63,6 +64,8 @@ public class SyncTokenService extends AbstractServiceWithBackgroundCheck impleme
protected ClusterSyncHistory clusterSyncHistory = new ClusterSyncHistory();
+ private final LogSilencer logSilencer = new LogSilencer(logger);
+
public static SyncTokenService testConstructorAndActivate(
DiscoveryLiteConfig commonsConfig,
ResourceResolverFactory resourceResolverFactory,
@@ -128,7 +131,7 @@ public class SyncTokenService extends AbstractServiceWithBackgroundCheck impleme
protected void syncToken(final BaseTopologyView view, final Runnable callback) {
- startBackgroundCheck("SyncTokenService", new BackgroundCheck() {
+ startBackgroundCheck("SyncTokenService-" + view.getLocalClusterSyncTokenId(), new BackgroundCheck() {
@Override
public boolean check() {
@@ -172,9 +175,11 @@ public class SyncTokenService extends AbstractServiceWithBackgroundCheck impleme
if (updateToken) {
syncTokens.put(slingId, syncTokenId);
resourceResolver.commit();
- logger.info("storeMySyncToken: stored syncToken of slingId="+slingId+" as="+syncTokenId);
+ logSilencer.infoOrDebug("storeMySyncToken-" + syncTokenId,
+ "storeMySyncToken: stored syncToken of slingId="+slingId+" as="+syncTokenId);
} else {
- logger.info("storeMySyncToken: syncToken was left unchanged for slingId="+slingId+" at="+syncTokenId);
+ logSilencer.infoOrDebug("storeMySyncToken-" + syncTokenId,
+ "storeMySyncToken: syncToken was left unchanged for slingId="+slingId+" at="+syncTokenId);
}
return true;
} catch (LoginException e) {
@@ -208,10 +213,12 @@ public class SyncTokenService extends AbstractServiceWithBackgroundCheck impleme
boolean success = true;
StringBuffer historyEntry = new StringBuffer();
for (InstanceDescription instance : view.getLocalInstance().getClusterView().getInstances()) {
- Object currentValue = syncTokens.get(instance.getSlingId());
+ String instanceSlingId = instance.getSlingId();
+ Object currentValue = syncTokens.get(instanceSlingId);
if (currentValue == null) {
String msg = "no syncToken yet of "+instance.getSlingId();
- logger.info("seenAllSyncTokens: " + msg);
+ logSilencer.infoOrDebug("seenAllSyncToken-" + syncToken + "-no-" + instanceSlingId,
+ "seenAllSyncTokens: " + msg);
if (historyEntry.length() != 0) {
historyEntry.append(",");
}
@@ -221,7 +228,8 @@ public class SyncTokenService extends AbstractServiceWithBackgroundCheck impleme
String msg = "syncToken of " + instance.getSlingId()
+ " is " + currentValue
+ " waiting for " + syncToken;
- logger.info("seenAllSyncTokens: " + msg);
+ logSilencer.infoOrDebug("seenAllSyncToken-" + syncToken + "-wait-"+instanceSlingId,
+ "seenAllSyncTokens: " + msg);
if (historyEntry.length() != 0) {
historyEntry.append(",");
}
@@ -230,15 +238,18 @@ public class SyncTokenService extends AbstractServiceWithBackgroundCheck impleme
}
}
if (!success) {
- logger.info("seenAllSyncTokens: not yet seen all expected syncTokens (see above for details)");
+ logSilencer.infoOrDebug("seenAllSyncToken-result-" + syncToken,
+ "seenAllSyncTokens: not yet seen all expected syncTokens (see above for details)");
clusterSyncHistory.addHistoryEntry(view, historyEntry.toString());
return false;
} else {
- clusterSyncHistory.addHistoryEntry(view, "seen all syncTokens");
+ clusterSyncHistory.addHistoryEntry(view,
+ "seen all syncTokens");
}
resourceResolver.commit();
- logger.info("seenAllSyncTokens: seen all syncTokens!");
+ logSilencer.infoOrDebug("seenAllSyncToken-result-" + syncToken,
+ "seenAllSyncTokens: seen all syncTokens!");
return true;
} catch (LoginException e) {
logger.error("seenAllSyncTokens: could not login: "+e, e);
diff --git a/src/main/java/org/apache/sling/discovery/commons/providers/spi/package-info.java b/src/main/java/org/apache/sling/discovery/commons/providers/spi/package-info.java
index df9d9f2..6d1dbe0 100644
--- a/src/main/java/org/apache/sling/discovery/commons/providers/spi/package-info.java
+++ b/src/main/java/org/apache/sling/discovery/commons/providers/spi/package-info.java
@@ -20,9 +20,9 @@
/**
* Provides an SPI for providers, used by discovery.commons.providers.impl
*
- * @version 1.0.0
+ * @version 1.1.0
*/
-@Version("1.0.0")
+@Version("1.1.0")
package org.apache.sling.discovery.commons.providers.spi;
import org.osgi.annotation.versioning.Version;
diff --git a/src/main/java/org/apache/sling/discovery/commons/providers/util/LogSilencer.java b/src/main/java/org/apache/sling/discovery/commons/providers/util/LogSilencer.java
new file mode 100644
index 0000000..9b9ee0f
--- /dev/null
+++ b/src/main/java/org/apache/sling/discovery/commons/providers/util/LogSilencer.java
@@ -0,0 +1,97 @@
+/*
+ * 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.discovery.commons.providers.util;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+
+/**
+ * Helper class to help reduce log.info output. It will avoid repetitive
+ * log.info calls and instead log future occurrences to log.debug
+ */
+public class LogSilencer {
+
+ private static final long DEFAULT_AUTO_RESET_DELAY_MINUTES = 10;
+
+ private final Logger logger;
+
+ private final Object syncObj = new Object();
+
+ private final long autoResetDelayMillis;
+
+ private Map<String, String> lastMsgPerCategory;
+
+ private long autoResetTime = 0;
+
+ public LogSilencer(Logger logger, long autoResetDelayMinutes) {
+ this.logger = logger;
+ if (autoResetDelayMinutes > 0) {
+ autoResetDelayMillis = TimeUnit.MINUTES.toMillis(autoResetDelayMinutes);
+ } else {
+ autoResetDelayMillis = 0;
+ }
+ }
+
+ public LogSilencer(Logger logger) {
+ this(logger, DEFAULT_AUTO_RESET_DELAY_MINUTES);
+ }
+
+ public void infoOrDebug(String category, String msg) {
+ final boolean doLogInfo;
+ synchronized (syncObj) {
+ if (autoResetTime == 0 || System.currentTimeMillis() > autoResetTime) {
+ reset();
+ }
+ if (lastMsgPerCategory == null) {
+ lastMsgPerCategory = new HashMap<>();
+ }
+ final String localLastMsg = lastMsgPerCategory.get(category);
+ if (localLastMsg == null || !localLastMsg.equals(msg)) {
+ doLogInfo = true;
+ lastMsgPerCategory.put(category, msg);
+ } else {
+ doLogInfo = false;
+ }
+ }
+ if (doLogInfo) {
+ logger.info("{} {}", msg, "(future identical logs go to debug)");
+ } else {
+ logger.debug(msg);
+ }
+ }
+
+ public void infoOrDebug(String msg) {
+ infoOrDebug(null, msg);
+ }
+
+ public void reset() {
+ synchronized (syncObj) {
+ lastMsgPerCategory = null;
+ if (autoResetDelayMillis == 0) {
+ autoResetTime = Long.MAX_VALUE;
+ } else {
+ autoResetTime = System.currentTimeMillis() + autoResetDelayMillis;
+ }
+ }
+ }
+
+}
diff --git a/src/main/java/org/apache/sling/discovery/commons/providers/util/package-info.java b/src/main/java/org/apache/sling/discovery/commons/providers/util/package-info.java
index b4778b2..cbe2750 100644
--- a/src/main/java/org/apache/sling/discovery/commons/providers/util/package-info.java
+++ b/src/main/java/org/apache/sling/discovery/commons/providers/util/package-info.java
@@ -20,9 +20,9 @@
/**
* Provides some static helpers for providers of the Discovery API.
*
- * @version 1.0.0
+ * @version 1.1.0
*/
-@Version("1.0.0")
+@Version("1.1.0")
package org.apache.sling.discovery.commons.providers.util;
import org.osgi.annotation.versioning.Version;
diff --git a/src/test/java/org/apache/sling/discovery/commons/providers/DummyTopologyView.java b/src/test/java/org/apache/sling/discovery/commons/providers/DummyTopologyView.java
index e12b24d..140b747 100644
--- a/src/test/java/org/apache/sling/discovery/commons/providers/DummyTopologyView.java
+++ b/src/test/java/org/apache/sling/discovery/commons/providers/DummyTopologyView.java
@@ -30,6 +30,7 @@ import java.util.UUID;
import org.apache.sling.discovery.ClusterView;
import org.apache.sling.discovery.InstanceDescription;
import org.apache.sling.discovery.InstanceFilter;
+import org.apache.sling.discovery.commons.providers.spi.LocalClusterView;
public class DummyTopologyView extends BaseTopologyView {
@@ -199,7 +200,15 @@ public class DummyTopologyView extends BaseTopologyView {
String clusterId = id.getClusterView().getId();
DefaultClusterView cluster = clusters.get(clusterId);
if (cluster==null) {
- cluster = new DefaultClusterView(clusterId);
+ final ClusterView origCluster = id.getClusterView();
+ if (origCluster instanceof LocalClusterView) {
+ final LocalClusterView localOrigCluster = (LocalClusterView) origCluster;
+ final LocalClusterView clonedCluster = new LocalClusterView(origCluster.getId(), localOrigCluster.getLocalClusterSyncTokenId());
+ clonedCluster.setPartiallyStartedClusterNodeIds(localOrigCluster.getPartiallyStartedClusterNodeIds());
+ cluster = clonedCluster;
+ } else {
+ cluster = new DefaultClusterView(clusterId);
+ }
clusters.put(clusterId, cluster);
}
DefaultInstanceDescription clone = clone(cluster, id);
diff --git a/src/test/java/org/apache/sling/discovery/commons/providers/base/TestViewStateManager.java b/src/test/java/org/apache/sling/discovery/commons/providers/base/TestViewStateManager.java
index 3b29202..c48a115 100644
--- a/src/test/java/org/apache/sling/discovery/commons/providers/base/TestViewStateManager.java
+++ b/src/test/java/org/apache/sling/discovery/commons/providers/base/TestViewStateManager.java
@@ -21,12 +21,15 @@ package org.apache.sling.discovery.commons.providers.base;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -43,6 +46,7 @@ import org.apache.sling.discovery.commons.providers.DefaultInstanceDescription;
import org.apache.sling.discovery.commons.providers.DummyTopologyView;
import org.apache.sling.discovery.commons.providers.EventHelper;
import org.apache.sling.discovery.commons.providers.spi.ClusterSyncService;
+import org.apache.sling.discovery.commons.providers.spi.LocalClusterView;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -716,4 +720,173 @@ public class TestViewStateManager {
assertFalse(mgr.onlyDiffersInProperties(view6));
}
+ @Test
+ public void testSuppression_withDelay() throws Exception {
+ doTestSuppression(true);
+ }
+
+ @Test
+ public void testSuppression_withoutDelay() throws Exception {
+ doTestSuppression(false);
+ }
+
+ private void doTestSuppression(boolean minEventDelayHelepr) throws Exception {
+ logger.info("testSuppression: start");
+
+ final AtomicReference<TopologyView> topologyRef = new AtomicReference<>();
+ if (minEventDelayHelepr) {
+ mgr.installMinEventDelayHandler(new DiscoveryService() {
+
+ @Override
+ public TopologyView getTopology() {
+ return topologyRef.get();
+ }
+ }, new DummyScheduler(), 1);
+ }
+
+ final String slingId1 = UUID.randomUUID().toString();
+ final String slingId2 = UUID.randomUUID().toString();
+ final String clusterId = UUID.randomUUID().toString();
+ final String syncToken1 = "s1";
+ final LocalClusterView cluster1 = new LocalClusterView(clusterId, syncToken1);
+ final DummyTopologyView view1 = new DummyTopologyView(syncToken1)
+ .addInstance(slingId1, cluster1, true, true);
+ final String syncToken4 = "s4";
+ final LocalClusterView cluster4 = new LocalClusterView(clusterId, syncToken4);
+ final DummyTopologyView view4 = new DummyTopologyView(syncToken4)
+ .addInstance(slingId1, cluster4, true, true)
+ .addInstance(slingId2, cluster4, false, false);
+
+ final DummyListener listener = new DummyListener();
+ mgr.bind(listener);
+ TestHelper.assertNoEvents(listener);
+ mgr.handleActivated();
+ TestHelper.assertNoEvents(listener);
+
+ logger.info("testSuppression: handleNewView(view1)");
+ mgr.handleNewView(view1);
+ topologyRef.set(view1);
+ assertEquals(0, mgr.waitForAsyncEvents(5000));
+ TopologyEvent initEvent = EventHelper.newInitEvent(view1);
+ assertEvents(listener, initEvent);
+
+ // for a view change to "go undetected" ie not trigger any topology change,
+ // the list of instances must remain the same
+ // (but the syncToken can differ and it can have suppressed clusterNodeIds)
+ for(int i = 0; i < 100; i++) {
+ final String syncToken2SameAsS1 = "s1";
+ final LocalClusterView cluster1Suppressed = new LocalClusterView(clusterId, syncToken2SameAsS1);
+ final DummyTopologyView view1Suppressed = new DummyTopologyView(syncToken2SameAsS1)
+ .addInstance(slingId1, cluster1Suppressed, true, true);
+ cluster1Suppressed.setPartiallyStartedClusterNodeIds(Arrays.asList(1));
+ logger.info("testSuppression: handleNewView(view2[a])");
+ mgr.handleNewView(view1Suppressed);
+ topologyRef.set(view1Suppressed);
+ assertEquals(0, mgr.waitForAsyncEvents(5000));
+ TestHelper.assertNoEvents(listener);
+ }
+ for(int i = 0; i < 100; i++) {
+ final String syncToken2Different = "s1Suppressed";
+ final LocalClusterView cluster1Suppressed = new LocalClusterView(clusterId, syncToken2Different);
+ final DummyTopologyView view1Suppressed = new DummyTopologyView(syncToken2Different)
+ .addInstance(slingId1, cluster1Suppressed, true, true);
+ cluster1Suppressed.setPartiallyStartedClusterNodeIds(Arrays.asList(1));
+ logger.info("testSuppression: handleNewView(view2[b])");
+ mgr.handleNewView(view1Suppressed);
+ topologyRef.set(view1Suppressed);
+ assertEquals(0, mgr.waitForAsyncEvents(5000));
+ TestHelper.assertNoEvents(listener);
+ }
+
+ logger.info("testSuppression: handleNewView(view4)");
+ mgr.handleNewView(view4);
+ topologyRef.set(view4);
+ assertEquals(0, mgr.waitForAsyncEvents(5000));
+ assertEvents(listener, EventHelper.newChangingEvent(view1), EventHelper.newChangedEvent(view1, view4));
+ }
+
+ @Test
+ public void testEqualsIgnoreSyncToken() throws Exception {
+ // no previous view, so returns false
+ assertFalse(mgr.equalsIgnoreSyncToken(null));
+
+ final String clusterId = UUID.randomUUID().toString();
+ final String syncToken1 = "s1";
+ final DummyTopologyView view1 = createTopology(clusterId, syncToken1, 1);
+ assertTrue(mgr.handleNewViewNonDelayed(view1));
+ try {
+ mgr.equalsIgnoreSyncToken(null);
+ fail("should have thrown a NPE");
+ } catch(RuntimeException e) {
+ // ok
+ }
+ assertTrue(mgr.equalsIgnoreSyncToken(view1));
+
+ DummyTopologyView view;
+ for(int i = 1; i < 10; i++) {
+ // same instances, same syncToken, no partiallyStartedInstances => true
+ view = createTopology(view1, syncToken1, 0);
+ assertTrue(mgr.equalsIgnoreSyncToken(view));
+ // same instances, same syncToken, with partiallyStartedInstances => true
+ addPartiallyStartedInstance(view, 1);
+ assertTrue(mgr.equalsIgnoreSyncToken(view));
+
+ // different instances, same syncToken, no partiallyStartedInstances => false
+ view = createTopology(view1, syncToken1, i);
+ assertFalse(mgr.equalsIgnoreSyncToken(view));
+ // different instances, same syncToken, with partiallyStartedInstances => false
+ addPartiallyStartedInstance(view, 1);
+ assertFalse(mgr.equalsIgnoreSyncToken(view));
+
+ // same instances, different syncToken, no partiallyStartedInstances => false
+ final String differentSyncToken = "s2";
+ view = createTopology(view1, differentSyncToken, 0);
+ assertFalse(mgr.equalsIgnoreSyncToken(view));
+ // same instances, different syncToken, with partiallyStartedInstances => true
+ view = createTopology(view1, differentSyncToken, 0);
+ addPartiallyStartedInstance(view, 1);
+ assertTrue(mgr.equalsIgnoreSyncToken(view));
+
+ // different instances, different syncToken, no partiallyStartedInstances => false
+ view = createTopology(view1, differentSyncToken, i);
+ assertFalse(mgr.equalsIgnoreSyncToken(view));
+ // different instances, different syncToken, with partiallyStartedInstances => false
+ view = createTopology(view1, differentSyncToken, i);
+ addPartiallyStartedInstance(view, 1);
+ assertFalse(mgr.equalsIgnoreSyncToken(view));
+ }
+ }
+
+ private void addPartiallyStartedInstance(TopologyView view, Integer... clusterNodeIds) {
+ LocalClusterView local = (LocalClusterView) view.getLocalInstance().getClusterView();
+ local.setPartiallyStartedClusterNodeIds(Arrays.asList(clusterNodeIds));
+ }
+
+ private DummyTopologyView createTopology(String clusterId, String syncToken, int numInstances) {
+ final LocalClusterView cluster = new LocalClusterView(clusterId, syncToken);
+ final DummyTopologyView view = new DummyTopologyView(syncToken);
+ for(int i = 0; i < numInstances; i++) {
+ view.addInstance(UUID.randomUUID().toString(), cluster, i == 0, i == 0);
+ }
+ return view;
+ }
+
+ private DummyTopologyView createTopology(DummyTopologyView base, String syncToken,
+ int numAdditionalInstances) {
+ return createTopology(base.getLocalInstance().getClusterView(), syncToken,
+ numAdditionalInstances);
+ }
+
+ private DummyTopologyView createTopology(ClusterView baseCluster, String syncToken,
+ int numAdditionalInstances) {
+ final LocalClusterView cluster = new LocalClusterView(baseCluster.getId(), syncToken);
+ final DummyTopologyView view2 = new DummyTopologyView(syncToken);
+ for (InstanceDescription inst : baseCluster.getInstances()) {
+ view2.addInstance(inst.getSlingId(), cluster, inst.isLeader(), inst.isLocal());
+ }
+ for (int i = 0; i < numAdditionalInstances; i++) {
+ view2.addInstance(UUID.randomUUID().toString(), cluster, false, false);
+ }
+ return view2;
+ }
}
\ No newline at end of file
diff --git a/src/test/java/org/apache/sling/discovery/commons/providers/spi/LocalClusterViewTest.java b/src/test/java/org/apache/sling/discovery/commons/providers/spi/LocalClusterViewTest.java
new file mode 100644
index 0000000..243f0db
--- /dev/null
+++ b/src/test/java/org/apache/sling/discovery/commons/providers/spi/LocalClusterViewTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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.discovery.commons.providers.spi;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class LocalClusterViewTest {
+
+ private final LocalClusterView view = new LocalClusterView("id", "token");
+
+ @Test
+ public void partiallyStarted() {
+ assertFalse(view.hasPartiallyStartedInstances());
+ assertFalse(view.isPartiallyStarted(42));
+
+ view.setPartiallyStartedClusterNodeIds(Collections.<Integer>emptyList());
+ assertFalse(view.hasPartiallyStartedInstances());
+ assertFalse(view.isPartiallyStarted(42));
+
+ view.setPartiallyStartedClusterNodeIds(Arrays.asList(1, 2, 3));
+ assertTrue(view.hasPartiallyStartedInstances());
+ assertFalse(view.isPartiallyStarted(42));
+ assertTrue(view.isPartiallyStarted(1));
+ assertFalse(view.isPartiallyStarted(null));
+ }
+}
diff --git a/src/test/java/org/apache/sling/discovery/commons/providers/spi/base/TestOakSyncTokenService.java b/src/test/java/org/apache/sling/discovery/commons/providers/spi/base/TestOakSyncTokenService.java
index ce8b31a..6f99021 100644
--- a/src/test/java/org/apache/sling/discovery/commons/providers/spi/base/TestOakSyncTokenService.java
+++ b/src/test/java/org/apache/sling/discovery/commons/providers/spi/base/TestOakSyncTokenService.java
@@ -24,6 +24,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.lang.reflect.Field;
+import java.util.Arrays;
import java.util.UUID;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -36,11 +37,11 @@ import org.apache.sling.discovery.commons.providers.ViewStateManager;
import org.apache.sling.discovery.commons.providers.base.DummyListener;
import org.apache.sling.discovery.commons.providers.base.TestHelper;
import org.apache.sling.discovery.commons.providers.base.ViewStateManagerFactory;
+import org.apache.sling.discovery.commons.providers.spi.LocalClusterView;
import org.apache.sling.discovery.commons.providers.spi.base.AbstractServiceWithBackgroundCheck.BackgroundCheckRunnable;
import org.apache.sling.jcr.api.SlingRepository;
import org.junit.After;
import org.junit.Before;
-import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -223,4 +224,85 @@ public class TestOakSyncTokenService {
Object backgroundCheckRunnable = field.get(idMapService);
return (BackgroundCheckRunnable) backgroundCheckRunnable;
}
+
+ @Test
+ public void testPartiallyStartedInstance() throws Exception {
+ logger.info("testPartiallyStartedInstance: start");
+ OakBacklogClusterSyncService cs = OakBacklogClusterSyncService.testConstructorAndActivate(new SimpleCommonsConfig(), idMapService1, new DummySlingSettingsService(slingId1), factory1);
+ Lock lock = new ReentrantLock();
+ ViewStateManager vsm = ViewStateManagerFactory.newViewStateManager(lock, cs);
+ DummyListener l = new DummyListener();
+ vsm.bind(l);
+ vsm.handleActivated();
+
+ final DummyTopologyView view1 = TestHelper.newView(true, slingId1, slingId1, slingId1);
+ {
+ // simulate a view with just itself (slingId1 / 1)
+ vsm.handleNewView(view1);
+ cs.triggerBackgroundCheck();
+ assertEquals(0, vsm.waitForAsyncEvents(1000));
+ assertEquals(0, l.countEvents());
+ DescriptorHelper.setDiscoveryLiteDescriptor(factory1, new DiscoveryLiteDescriptorBuilder().me(1).seq(1).activeIds(1).setFinal(true));
+ assertTrue(idMapService1.waitForInit(5000));
+ cs.triggerBackgroundCheck();
+ assertEquals(0, vsm.waitForAsyncEvents(1000));
+ assertEquals(1, l.countEvents());
+ }
+
+ assertTrue(idMapService1.waitForInit(5000));
+
+ {
+ // simulate a new instance coming up - first it will show up in oak leases/lite-view
+ DescriptorHelper.setDiscoveryLiteDescriptor(factory1, new DiscoveryLiteDescriptorBuilder().me(1).seq(2).activeIds(1, 2).setFinal(true));
+
+ // the view is still the same (only contains slingId1) - but it has the flag 'partial' set
+ final String syncToken2 = "s2";
+ final String clusterId = view1.getLocalInstance().getClusterView().getId();
+ final LocalClusterView cluster1Suppressed = new LocalClusterView(clusterId, syncToken2);
+ final DummyTopologyView view1Suppressed = new DummyTopologyView(syncToken2)
+ .addInstance(slingId1, cluster1Suppressed, true, true);
+ cluster1Suppressed.setPartiallyStartedClusterNodeIds(Arrays.asList(2));
+
+ vsm.handleNewView(view1Suppressed);
+ cs.triggerBackgroundCheck();
+ assertEquals(0, vsm.waitForAsyncEvents(1000));
+ assertEquals(1, l.countEvents());
+ }
+ final String slingId2 = UUID.randomUUID().toString();
+ {
+ // now define slingId for activeId == 2
+ IdMapService idMapService2 = IdMapService.testConstructor(
+ new SimpleCommonsConfig(), new DummySlingSettingsService(slingId2), factory2);
+ DescriptorHelper.setDiscoveryLiteDescriptor(factory2, new DiscoveryLiteDescriptorBuilder().setFinal(true).me(2).seq(2).activeIds(1, 2));
+ assertTrue(idMapService2.waitForInit(5000));
+ }
+ {
+ // now that shouldn't have triggered anything yet towards the listeners
+ final String syncToken2 = "s2";
+ final String clusterId = view1.getLocalInstance().getClusterView().getId();
+ final LocalClusterView cluster1Suppressed = new LocalClusterView(clusterId, syncToken2);
+ final DummyTopologyView view1Suppressed = new DummyTopologyView(syncToken2)
+ .addInstance(slingId1, cluster1Suppressed, true, true);
+ cluster1Suppressed.setPartiallyStartedClusterNodeIds(Arrays.asList(2));
+
+ vsm.handleNewView(view1Suppressed);
+ cs.triggerBackgroundCheck();
+ assertEquals(0, vsm.waitForAsyncEvents(1000));
+ assertEquals(1, l.countEvents());
+ }
+ {
+ // now let's finish slingId2 startup - only this should trigger CHANGING/CHANGED
+ final String syncToken2 = "s2";
+ final String clusterId = view1.getLocalInstance().getClusterView().getId();
+ final LocalClusterView cluster1Suppressed = new LocalClusterView(clusterId, syncToken2);
+ final DummyTopologyView view2 = new DummyTopologyView(syncToken2)
+ .addInstance(slingId1, cluster1Suppressed, true, true)
+ .addInstance(slingId2, cluster1Suppressed, false, false);
+ vsm.handleNewView(view2);
+ cs.triggerBackgroundCheck();
+ assertEquals(0, vsm.waitForAsyncEvents(1000));
+ assertEquals(3, l.countEvents());
+ }
+ logger.info("testPartiallyStartedInstance: end");
+ }
}