Mastership service provider API
[openflowplugin.git] / openflowplugin-impl / src / main / java / org / opendaylight / openflowplugin / impl / statistics / StatisticsContextImpl.java
index 1f1ab71e1b72c7a5e88b851c42dd407cc98202b9..d3b2e72cd9cb2be0878272f8cdf06e888a5a4275 100644 (file)
@@ -12,19 +12,16 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Function;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterators;
+import com.google.common.util.concurrent.AsyncFunction;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
 import io.netty.util.Timeout;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;
-import java.util.Optional;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
@@ -36,16 +33,15 @@ import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
-import org.opendaylight.openflowplugin.api.openflow.device.handlers.ClusterInitializationPhaseHandler;
 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState;
-import org.opendaylight.openflowplugin.api.openflow.lifecycle.MastershipChangeListener;
+import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipWatcher;
 import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager;
+import org.opendaylight.openflowplugin.impl.datastore.MultipartWriterProvider;
 import org.opendaylight.openflowplugin.impl.rpc.AbstractRequestContext;
 import org.opendaylight.openflowplugin.impl.rpc.listener.ItemLifecycleListenerImpl;
 import org.opendaylight.openflowplugin.impl.services.util.RequestContextUtil;
-import org.opendaylight.openflowplugin.impl.datastore.MultipartWriterProvider;
 import org.opendaylight.openflowplugin.impl.statistics.services.dedicated.StatisticsGatheringOnTheFlyService;
 import org.opendaylight.openflowplugin.impl.statistics.services.dedicated.StatisticsGatheringService;
 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
@@ -63,26 +59,22 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
     private final Collection<RequestContext<?>> requestContexts = new HashSet<>();
     private final DeviceContext deviceContext;
     private final DeviceState devState;
-    private final ListenableFuture<Boolean> emptyFuture;
     private final boolean isStatisticsPollingOn;
     private final Object collectionStatTypeLock = new Object();
     private final ConvertorExecutor convertorExecutor;
     private final MultipartWriterProvider statisticsWriterProvider;
+    private final DeviceInfo deviceInfo;
+    private final StatisticsManager myManager;
     @GuardedBy("collectionStatTypeLock")
     private List<MultipartType> collectingStatType;
-
     private StatisticsGatheringService<T> statisticsGatheringService;
     private StatisticsGatheringOnTheFlyService<T> statisticsGatheringOnTheFlyService;
     private Timeout pollTimeout;
-    private final DeviceInfo deviceInfo;
-    private final StatisticsManager myManager;
+    private ContextChainMastershipWatcher contextChainMastershipWatcher;
 
+    private volatile ContextState state = ContextState.INITIALIZATION;
     private volatile boolean schedulingEnabled;
-    private volatile CONTEXT_STATE state;
-    private ClusterInitializationPhaseHandler clusterInitializationPhaseHandler;
-    private ClusterInitializationPhaseHandler initialSubmitHandler;
-
-    private ListenableFuture<Boolean> lastDataGathering;
+    private volatile ListenableFuture<Boolean> lastDataGathering;
 
     StatisticsContextImpl(final boolean isStatisticsPollingOn,
                           @Nonnull final DeviceContext deviceContext,
@@ -93,13 +85,11 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
         this.devState = Preconditions.checkNotNull(deviceContext.getDeviceState());
         this.isStatisticsPollingOn = isStatisticsPollingOn;
         this.convertorExecutor = convertorExecutor;
-        emptyFuture = Futures.immediateFuture(false);
         statisticsGatheringService = new StatisticsGatheringService<>(this, deviceContext);
         statisticsGatheringOnTheFlyService = new StatisticsGatheringOnTheFlyService<>(this,
             deviceContext, convertorExecutor, statisticsWriterProvider);
         itemLifeCycleListener = new ItemLifecycleListenerImpl(deviceContext);
         statListForCollectingInitialization();
-        this.state = CONTEXT_STATE.INITIALIZATION;
         this.deviceInfo = deviceContext.getDeviceInfo();
         this.myManager = myManager;
         this.lastDataGathering = null;
@@ -130,46 +120,39 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
             if (devState.isQueueStatisticsAvailable()) {
                 statListForCollecting.add(MultipartType.OFPMPQUEUE);
             }
-            collectingStatType = ImmutableList.<MultipartType>copyOf(statListForCollecting);
+            collectingStatType = ImmutableList.copyOf(statListForCollecting);
         }
     }
 
-
-    @Override
-    public ListenableFuture<Boolean> initialGatherDynamicData() {
-        return gatherDynamicData(true);
-    }
-
     @Override
-    public ListenableFuture<Boolean> gatherDynamicData(){
-        return gatherDynamicData(false);
-    }
-
-    private ListenableFuture<Boolean> gatherDynamicData(final boolean initial) {
-        this.lastDataGathering = null;
+    public ListenableFuture<Boolean> gatherDynamicData() {
         if (!isStatisticsPollingOn) {
             LOG.debug("Statistics for device {} is not enabled.", getDeviceInfo().getNodeId().getValue());
             return Futures.immediateFuture(Boolean.TRUE);
         }
-        final ListenableFuture<Boolean> errorResultFuture = deviceConnectionCheck();
-        if (errorResultFuture != null) {
-            return errorResultFuture;
+
+        if (Objects.isNull(lastDataGathering)
+                || lastDataGathering.isCancelled()
+                || lastDataGathering.isDone()) {
+            lastDataGathering = Futures.immediateFuture(Boolean.TRUE);
         }
-        synchronized (collectionStatTypeLock) {
-            final Iterator<MultipartType> statIterator = collectingStatType.iterator();
-            final SettableFuture<Boolean> settableStatResultFuture = SettableFuture.create();
 
+        synchronized (collectionStatTypeLock) {
             // write start timestamp to state snapshot container
             StatisticsGatheringUtils.markDeviceStateSnapshotStart(deviceContext);
 
-            statChainFuture(statIterator, settableStatResultFuture, initial);
+            lastDataGathering = collectingStatType.stream().reduce(
+                    lastDataGathering,
+                    this::statChainFuture,
+                    (a, b) -> Futures.transformAsync(a, (AsyncFunction<Boolean, Boolean>) result -> b));
 
             // write end timestamp to state snapshot container
-            Futures.addCallback(settableStatResultFuture, new FutureCallback<Boolean>() {
+            Futures.addCallback(lastDataGathering, new FutureCallback<Boolean>() {
                 @Override
-                public void onSuccess(@Nullable final Boolean result) {
-                    StatisticsGatheringUtils.markDeviceStateSnapshotEnd(deviceContext, true);
+                public void onSuccess(final Boolean result) {
+                    StatisticsGatheringUtils.markDeviceStateSnapshotEnd(deviceContext, result);
                 }
+
                 @Override
                 public void onFailure(final Throwable t) {
                     if (!(t instanceof TransactionChainClosedException)) {
@@ -177,38 +160,38 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
                     }
                 }
             });
-            this.lastDataGathering = settableStatResultFuture;
-            return settableStatResultFuture;
         }
+
+        return lastDataGathering;
     }
 
-    private ListenableFuture<Boolean> chooseStat(final MultipartType multipartType, final boolean initial){
+    private ListenableFuture<Boolean> chooseStat(final MultipartType multipartType){
         ListenableFuture<Boolean> result = Futures.immediateCheckedFuture(Boolean.TRUE);
 
         switch (multipartType) {
             case OFPMPFLOW:
-                result = collectFlowStatistics(multipartType, initial);
+                result = collectStatistics(multipartType, devState.isFlowStatisticsAvailable(), true);
                 break;
             case OFPMPTABLE:
-                result = collectTableStatistics(multipartType);
+                result = collectStatistics(multipartType, devState.isTableStatisticsAvailable(), false);
                 break;
             case OFPMPPORTSTATS:
-                result = collectPortStatistics(multipartType);
+                result = collectStatistics(multipartType, devState.isPortStatisticsAvailable(), false);
                 break;
             case OFPMPQUEUE:
-                result = collectQueueStatistics(multipartType);
+                result = collectStatistics(multipartType, devState.isQueueStatisticsAvailable(), false);
                 break;
             case OFPMPGROUPDESC:
-                result = collectGroupDescStatistics(multipartType);
+                result = collectStatistics(multipartType, devState.isGroupAvailable(), false);
                 break;
             case OFPMPGROUP:
-                result = collectGroupStatistics(multipartType);
+                result = collectStatistics(multipartType, devState.isGroupAvailable(), false);
                 break;
             case OFPMPMETERCONFIG:
-                result = collectMeterConfigStatistics(multipartType);
+                result = collectStatistics(multipartType, devState.isMetersAvailable(), false);
                 break;
             case OFPMPMETER:
-                result = collectMeterStatistics(multipartType);
+                result = collectStatistics(multipartType, devState.isMetersAvailable(), false);
                 break;
             default:
                 LOG.warn("Unsupported Statistics type {}", multipartType);
@@ -232,22 +215,16 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
 
     @Override
     public void close() {
-        if (CONTEXT_STATE.TERMINATION.equals(getState())) {
+        if (ContextState.TERMINATION.equals(state)) {
             if (LOG.isDebugEnabled()) {
                 LOG.debug("StatisticsContext for node {} is already in TERMINATION state.", getDeviceInfo().getLOGValue());
             }
         } else {
-
-            this.state = CONTEXT_STATE.TERMINATION;
-
-            for (final Iterator<RequestContext<?>> iterator = Iterators.consumingIterator(requestContexts.iterator());
-                 iterator.hasNext(); ) {
-                RequestContextUtil.closeRequestContextWithRpcError(iterator.next(), CONNECTION_CLOSED);
-            }
-
-            if (null != pollTimeout && !pollTimeout.isExpired()) {
-                pollTimeout.cancel();
-            }
+            this.state = ContextState.TERMINATION;
+            stopGatheringData();
+            requestContexts.forEach(requestContext -> RequestContextUtil
+                    .closeRequestContextWithRpcError(requestContext, CONNECTION_CLOSED));
+            requestContexts.clear();
         }
     }
 
@@ -266,185 +243,44 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
         this.pollTimeout = pollTimeout;
     }
 
-    @Override
-    public Optional<Timeout> getPollTimeout() {
-        return Optional.ofNullable(pollTimeout);
-    }
-
-    private void statChainFuture(final Iterator<MultipartType> iterator, final SettableFuture<Boolean> resultFuture, final boolean initial) {
-        if (ConnectionContext.CONNECTION_STATE.RIP.equals(deviceContext.getPrimaryConnectionContext().getConnectionState())) {
-            final String errMsg = String.format("Device connection is closed for Node : %s.",
-                    getDeviceInfo().getNodeId());
-            LOG.debug(errMsg);
-            resultFuture.setException(new ConnectionException(errMsg));
-            return;
-        }
+    private ListenableFuture<Boolean> statChainFuture(final ListenableFuture<Boolean> prevFuture, final MultipartType multipartType) {
+        return Futures.transformAsync(deviceConnectionCheck(), (AsyncFunction<Boolean, Boolean>) connectionResult -> Futures
+                .transformAsync(prevFuture, (AsyncFunction<Boolean, Boolean>) result -> {
+                    LOG.debug("Status of previous stat iteration for node {}: {}", deviceInfo.getLOGValue(), result);
+                    LOG.debug("Stats iterating to next type for node {} of type {}",
+                            deviceInfo.getLOGValue(),
+                            multipartType);
 
-        if (!iterator.hasNext()) {
-            if (initial) {
-                Futures.addCallback(StatisticsGatheringUtils.gatherStatistics(
-                        statisticsGatheringService,
-                        getDeviceInfo(),
-                        MultipartType.OFPMPPORTDESC,
-                        deviceContext,
-                        deviceContext,
-                        false,
-                        convertorExecutor,
-                        statisticsWriterProvider), new FutureCallback<Boolean>() {
-                    @Override
-                    public void onSuccess(final Boolean result) {
-                        statChainFuture(iterator, resultFuture, false);
-                    }
-                    @Override
-                    public void onFailure(@Nonnull final Throwable t) {
-                        resultFuture.setException(t);
-                    }
-                });
-
-                return;
-            }
-
-            resultFuture.set(Boolean.TRUE);
-            LOG.debug("Stats collection successfully finished for node {}", getDeviceInfo().getLOGValue());
-            return;
-        }
-
-        final MultipartType nextType = iterator.next();
-        LOG.debug("Stats iterating to next type for node {} of type {}", getDeviceInfo().getLOGValue(), nextType);
-
-        final ListenableFuture<Boolean> deviceStatisticsCollectionFuture = chooseStat(nextType, initial);
-        Futures.addCallback(deviceStatisticsCollectionFuture, new FutureCallback<Boolean>() {
-            @Override
-            public void onSuccess(final Boolean result) {
-                statChainFuture(iterator, resultFuture, initial);
-            }
-            @Override
-            public void onFailure(@Nonnull final Throwable t) {
-                resultFuture.setException(t);
-            }
-        });
+                    return chooseStat(multipartType);
+                }));
     }
 
-    /**
-     * Method checks a device state. It returns null for be able continue. Otherwise it returns immediateFuture
-     * which has to be returned from caller too
-     *
-     * @return future
-     */
     @VisibleForTesting
     ListenableFuture<Boolean> deviceConnectionCheck() {
-        if (!ConnectionContext.CONNECTION_STATE.WORKING.equals(deviceContext.getPrimaryConnectionContext().getConnectionState())) {
-            ListenableFuture<Boolean> resultingFuture;
-            switch (deviceContext.getPrimaryConnectionContext().getConnectionState()) {
-                case RIP:
-                    final String errMsg = String.format("Device connection doesn't exist anymore. Primary connection status : %s",
-                        deviceContext.getPrimaryConnectionContext().getConnectionState());
-                    resultingFuture = Futures.immediateFailedFuture(new Throwable(errMsg));
-                    break;
-                default:
-                    resultingFuture = Futures.immediateCheckedFuture(Boolean.TRUE);
-                    break;
-            }
-            return resultingFuture;
-        }
-        return null;
-    }
-
-    //TODO: Refactor twice sending deviceContext into gatheringStatistics
-    private ListenableFuture<Boolean> collectFlowStatistics(final MultipartType multipartType, final boolean initial) {
-        return devState.isFlowStatisticsAvailable() ? StatisticsGatheringUtils.gatherStatistics(
-            statisticsGatheringOnTheFlyService,
-            getDeviceInfo(),
-            /*MultipartType.OFPMPFLOW*/ multipartType,
-            deviceContext,
-            deviceContext,
-            initial,
-            convertorExecutor,
-            statisticsWriterProvider) : emptyFuture;
-    }
-
-    private ListenableFuture<Boolean> collectTableStatistics(final MultipartType multipartType) {
-        return devState.isTableStatisticsAvailable() ? StatisticsGatheringUtils.gatherStatistics(
-            statisticsGatheringService,
-            getDeviceInfo(),
-                /*MultipartType.OFPMPTABLE*/ multipartType,
-            deviceContext,
-            deviceContext,
-            false,
-            convertorExecutor,
-            statisticsWriterProvider) : emptyFuture;
-    }
-
-    private ListenableFuture<Boolean> collectPortStatistics(final MultipartType multipartType) {
-        return devState.isPortStatisticsAvailable() ? StatisticsGatheringUtils.gatherStatistics(
-            statisticsGatheringService,
-            getDeviceInfo(),
-                /*MultipartType.OFPMPPORTSTATS*/ multipartType,
-            deviceContext,
-            deviceContext,
-            false,
-            convertorExecutor,
-            statisticsWriterProvider) : emptyFuture;
-    }
-
-    private ListenableFuture<Boolean> collectQueueStatistics(final MultipartType multipartType) {
-        return !devState.isQueueStatisticsAvailable() ? emptyFuture : StatisticsGatheringUtils.gatherStatistics(
-            statisticsGatheringService,
-            getDeviceInfo(),
-                /*MultipartType.OFPMPQUEUE*/ multipartType,
-            deviceContext,
-            deviceContext,
-            false,
-            convertorExecutor,
-            statisticsWriterProvider);
-    }
-
-    private ListenableFuture<Boolean> collectGroupDescStatistics(final MultipartType multipartType) {
-        return devState.isGroupAvailable() ? StatisticsGatheringUtils.gatherStatistics(
-            statisticsGatheringService,
-            getDeviceInfo(),
-                /*MultipartType.OFPMPGROUPDESC*/ multipartType,
-            deviceContext,
-            deviceContext,
-            false,
-            convertorExecutor,
-            statisticsWriterProvider) : emptyFuture;
-    }
+        if (ConnectionContext.CONNECTION_STATE.RIP.equals(deviceContext.getPrimaryConnectionContext().getConnectionState())) {
+            final String errMsg = String
+                    .format("Device connection for node %s doesn't exist anymore. Primary connection status : %s",
+                            getDeviceInfo().getNodeId(),
+                            deviceContext.getPrimaryConnectionContext().getConnectionState());
 
-    private ListenableFuture<Boolean> collectGroupStatistics(final MultipartType multipartType) {
-        return devState.isGroupAvailable() ? StatisticsGatheringUtils.gatherStatistics(
-            statisticsGatheringService,
-            getDeviceInfo(),
-                /*MultipartType.OFPMPGROUP*/ multipartType,
-            deviceContext,
-            deviceContext,
-            false,
-            convertorExecutor,
-            statisticsWriterProvider) : emptyFuture;
-    }
+            return Futures.immediateFailedFuture(new ConnectionException(errMsg));
+        }
 
-    private ListenableFuture<Boolean> collectMeterConfigStatistics(final MultipartType multipartType) {
-        return devState.isMetersAvailable() ? StatisticsGatheringUtils.gatherStatistics(
-            statisticsGatheringService,
-            getDeviceInfo(),
-                /*MultipartType.OFPMPMETERCONFIG*/ multipartType,
-            deviceContext,
-            deviceContext,
-            false,
-            convertorExecutor,
-            statisticsWriterProvider) : emptyFuture;
+        return Futures.immediateFuture(Boolean.TRUE);
     }
 
-    private ListenableFuture<Boolean> collectMeterStatistics(final MultipartType multipartType) {
-        return devState.isMetersAvailable() ? StatisticsGatheringUtils.gatherStatistics(
-            statisticsGatheringService,
-            getDeviceInfo(),
-                /*MultipartType.OFPMPMETER*/ multipartType,
-            deviceContext,
-            deviceContext,
-            false,
-            convertorExecutor,
-            statisticsWriterProvider) : emptyFuture;
+    private ListenableFuture<Boolean> collectStatistics(final MultipartType multipartType,
+                                                        final boolean supported,
+                                                        final boolean onTheFly) {
+        // TODO: Refactor twice sending deviceContext into gatheringStatistics
+        return supported ? StatisticsGatheringUtils.gatherStatistics(
+                onTheFly ? statisticsGatheringOnTheFlyService : statisticsGatheringService,
+                getDeviceInfo(),
+                multipartType,
+                deviceContext,
+                deviceContext,
+                convertorExecutor,
+                statisticsWriterProvider) : Futures.immediateFuture(Boolean.FALSE);
     }
 
     @VisibleForTesting
@@ -463,30 +299,23 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
     }
 
     @Override
-    public CONTEXT_STATE getState() {
-        return this.state;
-    }
-
-    @Override
-    public ServiceGroupIdentifier getServiceIdentifier() {
-        return this.deviceInfo.getServiceIdentifier();
+    public DeviceInfo getDeviceInfo() {
+        return this.deviceInfo;
     }
 
     @Override
-    public DeviceInfo getDeviceInfo() {
-        return this.deviceInfo;
+    public void registerMastershipWatcher(@Nonnull final ContextChainMastershipWatcher contextChainMastershipWatcher) {
+        this.contextChainMastershipWatcher = contextChainMastershipWatcher;
     }
 
     @Override
-    public ListenableFuture<Void> stopClusterServices() {
-        if (CONTEXT_STATE.TERMINATION.equals(this.state)) {
-            return Futures.immediateCancelledFuture();
-        }
+    public ListenableFuture<Void> closeServiceInstance() {
+        LOG.info("Stopping statistics context cluster services for node {}", deviceInfo.getLOGValue());
 
-        return Futures.transform(Futures.immediateFuture(null), new Function<Object, Void>() {
+        return Futures.transform(Futures.immediateFuture(null), new Function<Void, Void>() {
             @Nullable
             @Override
-            public Void apply(@Nullable Object input) {
+            public Void apply(@Nullable final Void input) {
                 schedulingEnabled = false;
                 stopGatheringData();
                 return null;
@@ -506,44 +335,41 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
 
     @Override
     public void stopGatheringData() {
-        if (Objects.nonNull(this.lastDataGathering)) {
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("Stop the running statistics gathering for node {}", this.deviceInfo.getLOGValue());
-            }
+        LOG.info("Stopping running statistics gathering for node {}", deviceInfo.getLOGValue());
 
+        if (Objects.nonNull(lastDataGathering) && !lastDataGathering.isDone() && !lastDataGathering.isCancelled()) {
             lastDataGathering.cancel(true);
         }
-    }
 
-    @Override
-    public void setLifecycleInitializationPhaseHandler(final ClusterInitializationPhaseHandler handler) {
-        this.clusterInitializationPhaseHandler = handler;
+        if (Objects.nonNull(pollTimeout) && !pollTimeout.isExpired()) {
+            pollTimeout.cancel();
+        }
     }
 
     @Override
-    public boolean onContextInstantiateService(final MastershipChangeListener mastershipChangeListener) {
-
+    public void instantiateServiceInstance() {
         LOG.info("Starting statistics context cluster services for node {}", deviceInfo.getLOGValue());
-
         this.statListForCollectingInitialization();
-        Futures.addCallback(this.initialGatherDynamicData(), new FutureCallback<Boolean>() {
 
+        Futures.addCallback(this.gatherDynamicData(), new FutureCallback<Boolean>() {
             @Override
             public void onSuccess(@Nullable Boolean aBoolean) {
-                mastershipChangeListener.onMasterRoleAcquired(
+                contextChainMastershipWatcher.onMasterRoleAcquired(
                         deviceInfo,
                         ContextChainMastershipState.INITIAL_GATHERING
                 );
-                if (initialSubmitHandler.initialSubmitTransaction()) {
-                    mastershipChangeListener.onMasterRoleAcquired(
+
+                if (deviceContext.initialSubmitTransaction()) {
+                    contextChainMastershipWatcher.onMasterRoleAcquired(
                             deviceInfo,
                             ContextChainMastershipState.INITIAL_SUBMIT
                     );
+
                     if (isStatisticsPollingOn) {
                         myManager.startScheduling(deviceInfo);
                     }
                 } else {
-                    mastershipChangeListener.onNotAbleToStartMastershipMandatory(
+                    contextChainMastershipWatcher.onNotAbleToStartMastershipMandatory(
                             deviceInfo,
                             "Initial transaction cannot be submitted."
                     );
@@ -552,18 +378,17 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
 
             @Override
             public void onFailure(@Nonnull Throwable throwable) {
-                mastershipChangeListener.onNotAbleToStartMastershipMandatory(
+                contextChainMastershipWatcher.onNotAbleToStartMastershipMandatory(
                         deviceInfo,
                         "Initial gathering statistics unsuccessful."
                 );
             }
         });
-
-        return this.clusterInitializationPhaseHandler.onContextInstantiateService(mastershipChangeListener);
     }
 
+    @Nonnull
     @Override
-    public void setInitialSubmitHandler(final ClusterInitializationPhaseHandler initialSubmitHandler) {
-        this.initialSubmitHandler = initialSubmitHandler;
+    public ServiceGroupIdentifier getIdentifier() {
+        return deviceInfo.getServiceIdentifier();
     }
 }