| /* |
| * 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.pulsar.broker.loadbalance.extensions.channel; |
| |
| import static java.lang.String.format; |
| import static java.util.concurrent.TimeUnit.MILLISECONDS; |
| import static java.util.concurrent.TimeUnit.SECONDS; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Assigning; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Deleted; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Free; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Init; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Owned; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Releasing; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Splitting; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.isActiveState; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.isInFlightState; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.ChannelState.Closed; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.ChannelState.Constructed; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.ChannelState.LeaderElectionServiceStarted; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.ChannelState.Started; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.EventType.Assign; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.EventType.Split; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.EventType.Unload; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.MetadataState.Jittery; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.MetadataState.Stable; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl.MetadataState.Unstable; |
| import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData.state; |
| import static org.apache.pulsar.common.naming.NamespaceName.SYSTEM_NAMESPACE; |
| import static org.apache.pulsar.common.topics.TopicCompactionStrategy.TABLE_VIEW_TAG; |
| import static org.apache.pulsar.metadata.api.extended.SessionEvent.SessionLost; |
| import static org.apache.pulsar.metadata.api.extended.SessionEvent.SessionReestablished; |
| import com.google.common.annotations.VisibleForTesting; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Optional; |
| import java.util.Set; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.Executor; |
| import java.util.concurrent.ScheduledFuture; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.TimeoutException; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.stream.Collectors; |
| import lombok.AllArgsConstructor; |
| import lombok.Getter; |
| import lombok.extern.slf4j.Slf4j; |
| import org.apache.commons.lang3.StringUtils; |
| import org.apache.commons.lang3.mutable.MutableInt; |
| import org.apache.commons.lang3.mutable.MutableObject; |
| import org.apache.pulsar.PulsarClusterMetadataSetup; |
| import org.apache.pulsar.broker.PulsarServerException; |
| import org.apache.pulsar.broker.PulsarService; |
| import org.apache.pulsar.broker.ServiceConfiguration; |
| import org.apache.pulsar.broker.loadbalance.LeaderElectionService; |
| import org.apache.pulsar.broker.loadbalance.extensions.BrokerRegistry; |
| import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl; |
| import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerWrapper; |
| import org.apache.pulsar.broker.loadbalance.extensions.LoadManagerContext; |
| import org.apache.pulsar.broker.loadbalance.extensions.manager.StateChangeListener; |
| import org.apache.pulsar.broker.loadbalance.extensions.models.Split; |
| import org.apache.pulsar.broker.loadbalance.extensions.models.Unload; |
| import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared; |
| import org.apache.pulsar.broker.namespace.NamespaceService; |
| import org.apache.pulsar.broker.service.BrokerServiceException; |
| import org.apache.pulsar.client.api.CompressionType; |
| import org.apache.pulsar.client.api.MessageId; |
| import org.apache.pulsar.client.api.Producer; |
| import org.apache.pulsar.client.api.PulsarClientException; |
| import org.apache.pulsar.client.api.Schema; |
| import org.apache.pulsar.client.api.TableView; |
| import org.apache.pulsar.common.naming.NamespaceBundle; |
| import org.apache.pulsar.common.naming.NamespaceBundleFactory; |
| import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm; |
| import org.apache.pulsar.common.naming.NamespaceBundles; |
| import org.apache.pulsar.common.naming.TopicDomain; |
| import org.apache.pulsar.common.naming.TopicName; |
| import org.apache.pulsar.common.stats.Metrics; |
| import org.apache.pulsar.common.topics.TopicCompactionStrategy; |
| import org.apache.pulsar.common.util.FutureUtil; |
| import org.apache.pulsar.metadata.api.MetadataStoreException; |
| import org.apache.pulsar.metadata.api.NotificationType; |
| import org.apache.pulsar.metadata.api.extended.SessionEvent; |
| |
| @Slf4j |
| public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel { |
| public static final String TOPIC = TopicName.get( |
| TopicDomain.persistent.value(), |
| SYSTEM_NAMESPACE, |
| "loadbalancer-service-unit-state").toString(); |
| |
| public static final CompressionType MSG_COMPRESSION_TYPE = CompressionType.ZSTD; |
| private static final long MAX_IN_FLIGHT_STATE_WAITING_TIME_IN_MILLIS = 30 * 1000; // 30sec |
| |
| private static final int OWNERSHIP_CLEAN_UP_MAX_WAIT_TIME_IN_MILLIS = 5000; |
| private static final int OWNERSHIP_CLEAN_UP_WAIT_RETRY_DELAY_IN_MILLIS = 100; |
| private static final int OWNERSHIP_CLEAN_UP_CONVERGENCE_DELAY_IN_MILLIS = 3000; |
| public static final long VERSION_ID_INIT = 1; // initial versionId |
| private static final long OWNERSHIP_MONITOR_DELAY_TIME_IN_SECS = 60; |
| public static final long MAX_CLEAN_UP_DELAY_TIME_IN_SECS = 3 * 60; // 3 mins |
| private static final long MIN_CLEAN_UP_DELAY_TIME_IN_SECS = 0; // 0 secs to clean immediately |
| private static final long MAX_CHANNEL_OWNER_ELECTION_WAITING_TIME_IN_SECS = 10; |
| private static final int MAX_OUTSTANDING_PUB_MESSAGES = 500; |
| private static final long MAX_OWNED_BUNDLE_COUNT_DELAY_TIME_IN_MILLIS = 10 * 60 * 1000; |
| private final PulsarService pulsar; |
| private final ServiceConfiguration config; |
| private final Schema<ServiceUnitStateData> schema; |
| private final Map<String, CompletableFuture<String>> getOwnerRequests; |
| private final String lookupServiceAddress; |
| private final Map<String, CompletableFuture<Void>> cleanupJobs; |
| private final StateChangeListeners stateChangeListeners; |
| private ExtensibleLoadManagerImpl loadManager; |
| private BrokerRegistry brokerRegistry; |
| private LeaderElectionService leaderElectionService; |
| private TableView<ServiceUnitStateData> tableview; |
| private Producer<ServiceUnitStateData> producer; |
| private ScheduledFuture<?> monitorTask; |
| private SessionEvent lastMetadataSessionEvent = SessionReestablished; |
| private long lastMetadataSessionEventTimestamp = 0; |
| private long inFlightStateWaitingTimeInMillis; |
| |
| private long ownershipMonitorDelayTimeInSecs; |
| private long semiTerminalStateWaitingTimeInMillis; |
| private long maxCleanupDelayTimeInSecs; |
| private long minCleanupDelayTimeInSecs; |
| // cleanup metrics |
| private long totalInactiveBrokerCleanupCnt = 0; |
| private long totalServiceUnitTombstoneCleanupCnt = 0; |
| |
| private long totalOrphanServiceUnitCleanupCnt = 0; |
| private AtomicLong totalCleanupErrorCnt = new AtomicLong(); |
| private long totalInactiveBrokerCleanupScheduledCnt = 0; |
| private long totalInactiveBrokerCleanupIgnoredCnt = 0; |
| private long totalInactiveBrokerCleanupCancelledCnt = 0; |
| private volatile ChannelState channelState; |
| private volatile long lastOwnEventHandledAt = 0; |
| private long lastOwnedServiceUnitCountAt = 0; |
| private int totalOwnedServiceUnitCnt = 0; |
| |
| public enum EventType { |
| Assign, |
| Split, |
| Unload, |
| Override |
| |
| } |
| |
| @Getter |
| @AllArgsConstructor |
| public static class Counters { |
| private final AtomicLong total; |
| private final AtomicLong failure; |
| public Counters(){ |
| total = new AtomicLong(); |
| failure = new AtomicLong(); |
| } |
| } |
| |
| // operation metrics |
| final Map<ServiceUnitState, Counters> ownerLookUpCounters; |
| final Map<EventType, Counters> eventCounters; |
| final Map<ServiceUnitState, Counters> handlerCounters; |
| |
| enum ChannelState { |
| Closed(0), |
| Constructed(1), |
| LeaderElectionServiceStarted(2), |
| Started(3); |
| |
| ChannelState(int id) { |
| this.id = id; |
| } |
| int id; |
| } |
| |
| enum MetadataState { |
| Stable, |
| Jittery, |
| Unstable |
| } |
| |
| public ServiceUnitStateChannelImpl(PulsarService pulsar) { |
| this.pulsar = pulsar; |
| this.config = pulsar.getConfig(); |
| this.lookupServiceAddress = pulsar.getLookupServiceAddress(); |
| this.schema = Schema.JSON(ServiceUnitStateData.class); |
| this.getOwnerRequests = new ConcurrentHashMap<>(); |
| this.cleanupJobs = new ConcurrentHashMap<>(); |
| this.stateChangeListeners = new StateChangeListeners(); |
| this.semiTerminalStateWaitingTimeInMillis = config.getLoadBalancerServiceUnitStateTombstoneDelayTimeInSeconds() |
| * 1000; |
| this.inFlightStateWaitingTimeInMillis = MAX_IN_FLIGHT_STATE_WAITING_TIME_IN_MILLIS; |
| this.ownershipMonitorDelayTimeInSecs = OWNERSHIP_MONITOR_DELAY_TIME_IN_SECS; |
| if (semiTerminalStateWaitingTimeInMillis < inFlightStateWaitingTimeInMillis) { |
| throw new IllegalArgumentException( |
| "Invalid Config: loadBalancerServiceUnitStateCleanUpDelayTimeInSeconds < " |
| + (MAX_IN_FLIGHT_STATE_WAITING_TIME_IN_MILLIS / 1000) + " secs"); |
| } |
| this.maxCleanupDelayTimeInSecs = MAX_CLEAN_UP_DELAY_TIME_IN_SECS; |
| this.minCleanupDelayTimeInSecs = MIN_CLEAN_UP_DELAY_TIME_IN_SECS; |
| |
| Map<ServiceUnitState, Counters> tmpOwnerLookUpCounters = new HashMap<>(); |
| Map<ServiceUnitState, Counters> tmpHandlerCounters = new HashMap<>(); |
| Map<EventType, Counters> tmpEventCounters = new HashMap<>(); |
| for (var state : ServiceUnitState.values()) { |
| tmpOwnerLookUpCounters.put(state, new Counters()); |
| tmpHandlerCounters.put(state, new Counters()); |
| } |
| for (var event : EventType.values()) { |
| tmpEventCounters.put(event, new Counters()); |
| } |
| ownerLookUpCounters = Map.copyOf(tmpOwnerLookUpCounters); |
| handlerCounters = Map.copyOf(tmpHandlerCounters); |
| eventCounters = Map.copyOf(tmpEventCounters); |
| this.channelState = Constructed; |
| } |
| |
| public void scheduleOwnershipMonitor() { |
| if (monitorTask == null) { |
| this.monitorTask = this.pulsar.getLoadManagerExecutor() |
| .scheduleWithFixedDelay(() -> { |
| try { |
| monitorOwnerships(brokerRegistry.getAvailableBrokersAsync() |
| .get(inFlightStateWaitingTimeInMillis, MILLISECONDS)); |
| } catch (Exception e) { |
| log.info("Failed to monitor the ownerships. will retry..", e); |
| } |
| }, |
| 0, ownershipMonitorDelayTimeInSecs, SECONDS); |
| log.info("This leader broker:{} started the ownership monitor.", |
| lookupServiceAddress); |
| } |
| } |
| |
| public void cancelOwnershipMonitor() { |
| if (monitorTask != null) { |
| monitorTask.cancel(false); |
| monitorTask = null; |
| log.info("This previous leader broker:{} stopped the ownership monitor.", |
| lookupServiceAddress); |
| } |
| } |
| |
| @Override |
| public void cleanOwnerships() { |
| doCleanup(lookupServiceAddress); |
| } |
| |
| public synchronized void start() throws PulsarServerException { |
| if (!validateChannelState(LeaderElectionServiceStarted, false)) { |
| throw new IllegalStateException("Invalid channel state:" + channelState.name()); |
| } |
| |
| boolean debug = debug(); |
| try { |
| this.brokerRegistry = getBrokerRegistry(); |
| this.brokerRegistry.addListener(this::handleBrokerRegistrationEvent); |
| this.leaderElectionService = getLeaderElectionService(); |
| var leader = leaderElectionService.readCurrentLeader().get( |
| MAX_CHANNEL_OWNER_ELECTION_WAITING_TIME_IN_SECS, TimeUnit.SECONDS); |
| if (leader.isPresent()) { |
| log.info("Successfully found the channel leader:{}.", leader.get()); |
| } else { |
| log.warn("Failed to find the channel leader."); |
| } |
| this.channelState = LeaderElectionServiceStarted; |
| loadManager = getLoadManager(); |
| |
| if (producer != null) { |
| producer.close(); |
| if (debug) { |
| log.info("Closed the channel producer."); |
| } |
| } |
| PulsarClusterMetadataSetup.createNamespaceIfAbsent |
| (pulsar.getPulsarResources(), SYSTEM_NAMESPACE, config.getClusterName()); |
| |
| ExtensibleLoadManagerImpl.createSystemTopic(pulsar, TOPIC); |
| |
| producer = pulsar.getClient().newProducer(schema) |
| .enableBatching(true) |
| .compressionType(MSG_COMPRESSION_TYPE) |
| .maxPendingMessages(MAX_OUTSTANDING_PUB_MESSAGES) |
| .blockIfQueueFull(true) |
| .topic(TOPIC) |
| .create(); |
| |
| if (debug) { |
| log.info("Successfully started the channel producer."); |
| } |
| |
| if (tableview != null) { |
| tableview.close(); |
| if (debug) { |
| log.info("Closed the channel tableview."); |
| } |
| } |
| tableview = pulsar.getClient().newTableViewBuilder(schema) |
| .topic(TOPIC) |
| .loadConf(Map.of( |
| "topicCompactionStrategyClassName", |
| ServiceUnitStateCompactionStrategy.class.getName())) |
| .create(); |
| tableview.listen((key, value) -> handle(key, value)); |
| var strategy = (ServiceUnitStateCompactionStrategy) TopicCompactionStrategy.getInstance(TABLE_VIEW_TAG); |
| if (strategy == null) { |
| String err = TABLE_VIEW_TAG + "tag TopicCompactionStrategy is null."; |
| log.error(err); |
| throw new IllegalStateException(err); |
| } |
| strategy.setSkippedMsgHandler((key, value) -> handleSkippedEvent(key)); |
| if (debug) { |
| log.info("Successfully started the channel tableview."); |
| } |
| pulsar.getLocalMetadataStore().registerSessionListener(this::handleMetadataSessionEvent); |
| if (debug) { |
| log.info("Successfully registered the handleMetadataSessionEvent"); |
| } |
| |
| channelState = Started; |
| log.info("Successfully started the channel."); |
| } catch (Exception e) { |
| String msg = "Failed to start the channel."; |
| log.error(msg, e); |
| throw new PulsarServerException(msg, e); |
| } |
| } |
| |
| @VisibleForTesting |
| protected BrokerRegistry getBrokerRegistry() { |
| return ((ExtensibleLoadManagerWrapper) pulsar.getLoadManager().get()) |
| .get().getBrokerRegistry(); |
| } |
| |
| @VisibleForTesting |
| protected LoadManagerContext getContext() { |
| return ((ExtensibleLoadManagerWrapper) pulsar.getLoadManager().get()) |
| .get().getContext(); |
| } |
| |
| @VisibleForTesting |
| protected ExtensibleLoadManagerImpl getLoadManager() { |
| return ExtensibleLoadManagerImpl.get(pulsar.getLoadManager().get()); |
| } |
| |
| @VisibleForTesting |
| protected LeaderElectionService getLeaderElectionService() { |
| return ((ExtensibleLoadManagerWrapper) pulsar.getLoadManager().get()) |
| .get().getLeaderElectionService(); |
| } |
| |
| public synchronized void close() throws PulsarServerException { |
| channelState = Closed; |
| boolean debug = debug(); |
| try { |
| leaderElectionService = null; |
| if (tableview != null) { |
| tableview.close(); |
| tableview = null; |
| if (debug) { |
| log.info("Successfully closed the channel tableview."); |
| } |
| } |
| |
| if (producer != null) { |
| producer.close(); |
| producer = null; |
| log.info("Successfully closed the channel producer."); |
| } |
| |
| if (brokerRegistry != null) { |
| brokerRegistry = null; |
| } |
| |
| if (monitorTask != null) { |
| monitorTask.cancel(true); |
| monitorTask = null; |
| log.info("Successfully cancelled the cleanup tasks"); |
| } |
| |
| if (stateChangeListeners != null) { |
| stateChangeListeners.close(); |
| } |
| |
| log.info("Successfully closed the channel."); |
| |
| } catch (Exception e) { |
| String msg = "Failed to close the channel."; |
| log.error(msg, e); |
| throw new PulsarServerException(msg, e); |
| } |
| } |
| |
| private boolean validateChannelState(ChannelState targetState, boolean checkLowerIds) { |
| int order = checkLowerIds ? -1 : 1; |
| if (Integer.compare(channelState.id, targetState.id) * order > 0) { |
| return false; |
| } |
| return true; |
| } |
| |
| private boolean debug() { |
| return ExtensibleLoadManagerImpl.debug(config, log); |
| } |
| |
| public CompletableFuture<Optional<String>> getChannelOwnerAsync() { |
| if (!validateChannelState(LeaderElectionServiceStarted, true)) { |
| return CompletableFuture.failedFuture( |
| new IllegalStateException("Invalid channel state:" + channelState.name())); |
| } |
| |
| return leaderElectionService.readCurrentLeader().thenApply(leader -> { |
| //expecting http://broker-xyz:port |
| // TODO: discard this protocol prefix removal |
| // by a util func that returns lookupServiceAddress(serviceUrl) |
| if (leader.isPresent()) { |
| String broker = leader.get().getServiceUrl(); |
| broker = broker.substring(broker.lastIndexOf('/') + 1); |
| return Optional.of(broker); |
| } else { |
| return Optional.empty(); |
| } |
| } |
| ); |
| } |
| |
| public CompletableFuture<Boolean> isChannelOwnerAsync() { |
| return getChannelOwnerAsync().thenApply(owner -> { |
| if (owner.isPresent()) { |
| return isTargetBroker(owner.get()); |
| } else { |
| String msg = "There is no channel owner now."; |
| log.error(msg); |
| throw new IllegalStateException(msg); |
| } |
| }); |
| } |
| |
| public boolean isChannelOwner() { |
| try { |
| return isChannelOwnerAsync().get( |
| MAX_CHANNEL_OWNER_ELECTION_WAITING_TIME_IN_SECS, TimeUnit.SECONDS); |
| } catch (InterruptedException | ExecutionException | TimeoutException e) { |
| String msg = "Failed to get the channel owner."; |
| log.error(msg, e); |
| throw new RuntimeException(msg, e); |
| } |
| } |
| |
| public boolean isOwner(String serviceUnit, String targetBroker) { |
| if (!validateChannelState(Started, true)) { |
| throw new IllegalStateException("Invalid channel state:" + channelState.name()); |
| } |
| var ownerFuture = getOwnerAsync(serviceUnit); |
| if (!ownerFuture.isDone() || ownerFuture.isCompletedExceptionally() || ownerFuture.isCancelled()) { |
| return false; |
| } |
| var owner = ownerFuture.join(); |
| if (owner.isPresent() && StringUtils.equals(targetBroker, owner.get())) { |
| return true; |
| } |
| return false; |
| } |
| |
| public boolean isOwner(String serviceUnit) { |
| return isOwner(serviceUnit, lookupServiceAddress); |
| } |
| |
| public CompletableFuture<Optional<String>> getOwnerAsync(String serviceUnit) { |
| if (!validateChannelState(Started, true)) { |
| return CompletableFuture.failedFuture( |
| new IllegalStateException("Invalid channel state:" + channelState.name())); |
| } |
| |
| ServiceUnitStateData data = tableview.get(serviceUnit); |
| ServiceUnitState state = state(data); |
| ownerLookUpCounters.get(state).getTotal().incrementAndGet(); |
| switch (state) { |
| case Owned -> { |
| return CompletableFuture.completedFuture(Optional.of(data.dstBroker())); |
| } |
| case Splitting -> { |
| return CompletableFuture.completedFuture(Optional.of(data.sourceBroker())); |
| } |
| case Assigning, Releasing -> { |
| return deferGetOwnerRequest(serviceUnit).whenComplete((__, e) -> { |
| if (e != null) { |
| ownerLookUpCounters.get(state).getFailure().incrementAndGet(); |
| } |
| }).thenApply( |
| broker -> broker == null ? Optional.empty() : Optional.of(broker)); |
| } |
| case Init, Free -> { |
| return CompletableFuture.completedFuture(Optional.empty()); |
| } |
| case Deleted -> { |
| ownerLookUpCounters.get(state).getFailure().incrementAndGet(); |
| return CompletableFuture.failedFuture(new IllegalArgumentException(serviceUnit + " is deleted.")); |
| } |
| default -> { |
| ownerLookUpCounters.get(state).getFailure().incrementAndGet(); |
| String errorMsg = String.format("Failed to process service unit state data: %s when get owner.", data); |
| log.error(errorMsg); |
| return CompletableFuture.failedFuture(new IllegalStateException(errorMsg)); |
| } |
| } |
| } |
| |
| private long getNextVersionId(String serviceUnit) { |
| var data = tableview.get(serviceUnit); |
| return getNextVersionId(data); |
| } |
| |
| private long getNextVersionId(ServiceUnitStateData data) { |
| return data == null ? VERSION_ID_INIT : data.versionId() + 1; |
| } |
| |
| public CompletableFuture<String> publishAssignEventAsync(String serviceUnit, String broker) { |
| if (!validateChannelState(Started, true)) { |
| return CompletableFuture.failedFuture( |
| new IllegalStateException("Invalid channel state:" + channelState.name())); |
| } |
| EventType eventType = Assign; |
| eventCounters.get(eventType).getTotal().incrementAndGet(); |
| CompletableFuture<String> getOwnerRequest = deferGetOwnerRequest(serviceUnit); |
| |
| pubAsync(serviceUnit, new ServiceUnitStateData(Assigning, broker, getNextVersionId(serviceUnit))) |
| .whenComplete((__, ex) -> { |
| if (ex != null) { |
| getOwnerRequests.remove(serviceUnit, getOwnerRequest); |
| if (!getOwnerRequest.isCompletedExceptionally()) { |
| getOwnerRequest.completeExceptionally(ex); |
| } |
| eventCounters.get(eventType).getFailure().incrementAndGet(); |
| } |
| }); |
| return getOwnerRequest; |
| } |
| |
| private CompletableFuture<Void> publishOverrideEventAsync(String serviceUnit, |
| ServiceUnitStateData orphanData, |
| ServiceUnitStateData override) { |
| if (!validateChannelState(Started, true)) { |
| throw new IllegalStateException("Invalid channel state:" + channelState.name()); |
| } |
| EventType eventType = EventType.Override; |
| eventCounters.get(eventType).getTotal().incrementAndGet(); |
| return pubAsync(serviceUnit, override).whenComplete((__, e) -> { |
| if (e != null) { |
| eventCounters.get(eventType).getFailure().incrementAndGet(); |
| log.error("Failed to override serviceUnit:{} from orphanData:{} to overrideData:{}", |
| serviceUnit, orphanData, override, e); |
| } |
| }).thenApply(__ -> null); |
| } |
| |
| public CompletableFuture<Void> publishUnloadEventAsync(Unload unload) { |
| if (!validateChannelState(Started, true)) { |
| return CompletableFuture.failedFuture( |
| new IllegalStateException("Invalid channel state:" + channelState.name())); |
| } |
| EventType eventType = Unload; |
| eventCounters.get(eventType).getTotal().incrementAndGet(); |
| String serviceUnit = unload.serviceUnit(); |
| ServiceUnitStateData next; |
| if (isTransferCommand(unload)) { |
| next = new ServiceUnitStateData( |
| Releasing, unload.destBroker().get(), unload.sourceBroker(), |
| unload.force(), getNextVersionId(serviceUnit)); |
| } else { |
| next = new ServiceUnitStateData( |
| Releasing, null, unload.sourceBroker(), unload.force(), getNextVersionId(serviceUnit)); |
| } |
| return pubAsync(serviceUnit, next).whenComplete((__, ex) -> { |
| if (ex != null) { |
| eventCounters.get(eventType).getFailure().incrementAndGet(); |
| } |
| }).thenApply(__ -> null); |
| } |
| |
| public CompletableFuture<Void> publishSplitEventAsync(Split split) { |
| if (!validateChannelState(Started, true)) { |
| return CompletableFuture.failedFuture( |
| new IllegalStateException("Invalid channel state:" + channelState.name())); |
| } |
| EventType eventType = Split; |
| eventCounters.get(eventType).getTotal().incrementAndGet(); |
| String serviceUnit = split.serviceUnit(); |
| ServiceUnitStateData next = |
| new ServiceUnitStateData(Splitting, null, split.sourceBroker(), |
| split.splitServiceUnitToDestBroker(), getNextVersionId(serviceUnit)); |
| return pubAsync(serviceUnit, next).whenComplete((__, ex) -> { |
| if (ex != null) { |
| eventCounters.get(eventType).getFailure().incrementAndGet(); |
| } |
| }).thenApply(__ -> null); |
| } |
| |
| private void handle(String serviceUnit, ServiceUnitStateData data) { |
| long totalHandledRequests = getHandlerTotalCounter(data).incrementAndGet(); |
| if (debug()) { |
| log.info("{} received a handle request for serviceUnit:{}, data:{}. totalHandledRequests:{}", |
| lookupServiceAddress, serviceUnit, data, totalHandledRequests); |
| } |
| |
| ServiceUnitState state = state(data); |
| try { |
| switch (state) { |
| case Owned -> handleOwnEvent(serviceUnit, data); |
| case Assigning -> handleAssignEvent(serviceUnit, data); |
| case Releasing -> handleReleaseEvent(serviceUnit, data); |
| case Splitting -> handleSplitEvent(serviceUnit, data); |
| case Deleted -> handleDeleteEvent(serviceUnit, data); |
| case Free -> handleFreeEvent(serviceUnit, data); |
| case Init -> handleInitEvent(serviceUnit); |
| default -> throw new IllegalStateException("Failed to handle channel data:" + data); |
| } |
| } catch (Throwable e){ |
| log.error("Failed to handle the event. serviceUnit:{}, data:{}, handlerFailureCount:{}", |
| serviceUnit, data, getHandlerFailureCounter(data).incrementAndGet(), e); |
| throw e; |
| } |
| } |
| |
| private static boolean isTransferCommand(ServiceUnitStateData data) { |
| if (data == null) { |
| return false; |
| } |
| return StringUtils.isNotEmpty(data.dstBroker()) |
| && StringUtils.isNotEmpty(data.sourceBroker()); |
| } |
| |
| private static boolean isTransferCommand(Unload data) { |
| return data.destBroker().isPresent(); |
| } |
| |
| private static String getLogEventTag(ServiceUnitStateData data) { |
| return data == null ? Init.toString() : |
| isTransferCommand(data) ? "Transfer:" + data.state() : data.state().toString(); |
| } |
| |
| private AtomicLong getHandlerTotalCounter(ServiceUnitStateData data) { |
| return getHandlerCounter(data, true); |
| } |
| |
| private AtomicLong getHandlerFailureCounter(ServiceUnitStateData data) { |
| return getHandlerCounter(data, false); |
| } |
| |
| private AtomicLong getHandlerCounter(ServiceUnitStateData data, boolean total) { |
| var state = state(data); |
| var counter = total |
| ? handlerCounters.get(state).getTotal() : handlerCounters.get(state).getFailure(); |
| if (counter == null) { |
| throw new IllegalStateException("Unknown state:" + state); |
| } |
| return counter; |
| } |
| |
| private void log(Throwable e, String serviceUnit, ServiceUnitStateData data, ServiceUnitStateData next) { |
| if (e == null) { |
| if (log.isDebugEnabled() || isTransferCommand(data)) { |
| long handlerTotalCount = getHandlerTotalCounter(data).get(); |
| long handlerFailureCount = getHandlerFailureCounter(data).get(); |
| log.info("{} handled {} event for serviceUnit:{}, cur:{}, next:{}, " |
| + "totalHandledRequests:{}, totalFailedRequests:{}", |
| lookupServiceAddress, getLogEventTag(data), serviceUnit, |
| data == null ? "" : data, |
| next == null ? "" : next, |
| handlerTotalCount, handlerFailureCount |
| ); |
| } |
| } else { |
| long handlerTotalCount = getHandlerTotalCounter(data).get(); |
| long handlerFailureCount = getHandlerFailureCounter(data).incrementAndGet(); |
| log.error("{} failed to handle {} event for serviceUnit:{}, cur:{}, next:{}, " |
| + "totalHandledRequests:{}, totalFailedRequests:{}", |
| lookupServiceAddress, getLogEventTag(data), serviceUnit, |
| data == null ? "" : data, |
| next == null ? "" : next, |
| handlerTotalCount, handlerFailureCount, |
| e); |
| } |
| } |
| |
| private void handleSkippedEvent(String serviceUnit) { |
| var getOwnerRequest = getOwnerRequests.get(serviceUnit); |
| if (getOwnerRequest != null) { |
| var data = tableview.get(serviceUnit); |
| if (data != null && data.state() == Owned) { |
| getOwnerRequest.complete(data.dstBroker()); |
| getOwnerRequests.remove(serviceUnit); |
| stateChangeListeners.notify(serviceUnit, data, null); |
| } |
| } |
| } |
| |
| private void handleOwnEvent(String serviceUnit, ServiceUnitStateData data) { |
| var getOwnerRequest = getOwnerRequests.remove(serviceUnit); |
| if (getOwnerRequest != null) { |
| getOwnerRequest.complete(data.dstBroker()); |
| } |
| stateChangeListeners.notify(serviceUnit, data, null); |
| if (isTargetBroker(data.dstBroker())) { |
| log(null, serviceUnit, data, null); |
| pulsar.getNamespaceService() |
| .onNamespaceBundleOwned(LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit)); |
| lastOwnEventHandledAt = System.currentTimeMillis(); |
| } else if (data.force() && isTargetBroker(data.sourceBroker())) { |
| closeServiceUnit(serviceUnit); |
| } |
| } |
| |
| private void handleAssignEvent(String serviceUnit, ServiceUnitStateData data) { |
| if (isTargetBroker(data.dstBroker())) { |
| ServiceUnitStateData next = new ServiceUnitStateData( |
| Owned, data.dstBroker(), data.sourceBroker(), getNextVersionId(data)); |
| stateChangeListeners.notifyOnCompletion(pubAsync(serviceUnit, next), serviceUnit, data) |
| .whenComplete((__, e) -> log(e, serviceUnit, data, next)); |
| } |
| } |
| |
| private void handleReleaseEvent(String serviceUnit, ServiceUnitStateData data) { |
| if (isTargetBroker(data.sourceBroker())) { |
| ServiceUnitStateData next; |
| if (isTransferCommand(data)) { |
| next = new ServiceUnitStateData( |
| Assigning, data.dstBroker(), data.sourceBroker(), getNextVersionId(data)); |
| // TODO: when close, pass message to clients to connect to the new broker |
| } else { |
| next = new ServiceUnitStateData( |
| Free, null, data.sourceBroker(), getNextVersionId(data)); |
| } |
| stateChangeListeners.notifyOnCompletion(closeServiceUnit(serviceUnit) |
| .thenCompose(__ -> pubAsync(serviceUnit, next)), serviceUnit, data) |
| .whenComplete((__, e) -> log(e, serviceUnit, data, next)); |
| } |
| } |
| |
| private void handleSplitEvent(String serviceUnit, ServiceUnitStateData data) { |
| if (isTargetBroker(data.sourceBroker())) { |
| stateChangeListeners.notifyOnCompletion(splitServiceUnit(serviceUnit, data), serviceUnit, data) |
| .whenComplete((__, e) -> log(e, serviceUnit, data, null)); |
| } |
| } |
| |
| private void handleFreeEvent(String serviceUnit, ServiceUnitStateData data) { |
| var getOwnerRequest = getOwnerRequests.remove(serviceUnit); |
| if (getOwnerRequest != null) { |
| getOwnerRequest.complete(null); |
| } |
| stateChangeListeners.notify(serviceUnit, data, null); |
| if (isTargetBroker(data.sourceBroker())) { |
| log(null, serviceUnit, data, null); |
| } |
| } |
| |
| private void handleDeleteEvent(String serviceUnit, ServiceUnitStateData data) { |
| var getOwnerRequest = getOwnerRequests.remove(serviceUnit); |
| if (getOwnerRequest != null) { |
| getOwnerRequest.completeExceptionally(new IllegalStateException(serviceUnit + "has been deleted.")); |
| } |
| stateChangeListeners.notify(serviceUnit, data, null); |
| if (isTargetBroker(data.sourceBroker())) { |
| log(null, serviceUnit, data, null); |
| } |
| } |
| |
| private void handleInitEvent(String serviceUnit) { |
| var getOwnerRequest = getOwnerRequests.remove(serviceUnit); |
| if (getOwnerRequest != null) { |
| getOwnerRequest.complete(null); |
| } |
| stateChangeListeners.notify(serviceUnit, null, null); |
| log(null, serviceUnit, null, null); |
| } |
| |
| private CompletableFuture<MessageId> pubAsync(String serviceUnit, ServiceUnitStateData data) { |
| CompletableFuture<MessageId> future = new CompletableFuture<>(); |
| producer.newMessage() |
| .key(serviceUnit) |
| .value(data) |
| .sendAsync() |
| .whenComplete((messageId, e) -> { |
| if (e != null) { |
| log.error("Failed to publish the message: serviceUnit:{}, data:{}", |
| serviceUnit, data, e); |
| future.completeExceptionally(e); |
| } else { |
| future.complete(messageId); |
| } |
| }); |
| return future; |
| } |
| |
| private CompletableFuture<MessageId> tombstoneAsync(String serviceUnit) { |
| return pubAsync(serviceUnit, null); |
| } |
| |
| private boolean isTargetBroker(String broker) { |
| if (broker == null) { |
| return false; |
| } |
| return broker.equals(lookupServiceAddress); |
| } |
| |
| private CompletableFuture<String> deferGetOwnerRequest(String serviceUnit) { |
| var requested = new MutableObject<CompletableFuture<String>>(); |
| try { |
| return getOwnerRequests |
| .computeIfAbsent(serviceUnit, k -> { |
| CompletableFuture<String> future = new CompletableFuture<>(); |
| requested.setValue(future); |
| return future; |
| }); |
| } finally { |
| var future = requested.getValue(); |
| if (future != null) { |
| future.orTimeout(inFlightStateWaitingTimeInMillis, TimeUnit.MILLISECONDS) |
| .whenComplete((v, e) -> { |
| if (e != null) { |
| getOwnerRequests.remove(serviceUnit, future); |
| log.warn("Failed to getOwner for serviceUnit:{}", |
| serviceUnit, e); |
| } |
| } |
| ); |
| } |
| } |
| } |
| |
| private CompletableFuture<Integer> closeServiceUnit(String serviceUnit) { |
| long startTime = System.nanoTime(); |
| MutableInt unloadedTopics = new MutableInt(); |
| NamespaceBundle bundle = LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit); |
| return pulsar.getBrokerService().unloadServiceUnit( |
| bundle, |
| true, |
| pulsar.getConfig().getNamespaceBundleUnloadingTimeoutMs(), |
| TimeUnit.MILLISECONDS) |
| .thenApply(numUnloadedTopics -> { |
| unloadedTopics.setValue(numUnloadedTopics); |
| return numUnloadedTopics; |
| }) |
| .whenComplete((__, ex) -> { |
| // clean up topics that failed to unload from the broker ownership cache |
| pulsar.getBrokerService().cleanUnloadedTopicFromCache(bundle); |
| pulsar.getNamespaceService().onNamespaceBundleUnload(bundle); |
| double unloadBundleTime = TimeUnit.NANOSECONDS |
| .toMillis((System.nanoTime() - startTime)); |
| if (ex != null) { |
| log.error("Failed to close topics under bundle:{} in {} ms", |
| bundle.toString(), unloadBundleTime, ex); |
| } else { |
| log.info("Unloading bundle:{} with {} topics completed in {} ms", |
| bundle, unloadedTopics, unloadBundleTime); |
| } |
| }); |
| } |
| |
| private CompletableFuture<Void> splitServiceUnit(String serviceUnit, ServiceUnitStateData data) { |
| // Write the child ownerships to channel. |
| long startTime = System.nanoTime(); |
| NamespaceService namespaceService = pulsar.getNamespaceService(); |
| NamespaceBundleFactory bundleFactory = namespaceService.getNamespaceBundleFactory(); |
| NamespaceBundle bundle = LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit); |
| CompletableFuture<Void> completionFuture = new CompletableFuture<>(); |
| Map<String, Optional<String>> bundleToDestBroker = data.splitServiceUnitToDestBroker(); |
| List<Long> boundaries = null; |
| NamespaceBundleSplitAlgorithm nsBundleSplitAlgorithm = |
| namespaceService.getNamespaceBundleSplitAlgorithmByName( |
| config.getDefaultNamespaceBundleSplitAlgorithm()); |
| if (bundleToDestBroker != null && bundleToDestBroker.size() == 2) { |
| Set<Long> boundariesSet = new HashSet<>(); |
| String namespace = bundle.getNamespaceObject().toString(); |
| bundleToDestBroker.forEach((bundleRange, destBroker) -> { |
| NamespaceBundle subBundle = bundleFactory.getBundle(namespace, bundleRange); |
| boundariesSet.add(subBundle.getKeyRange().lowerEndpoint()); |
| boundariesSet.add(subBundle.getKeyRange().upperEndpoint()); |
| }); |
| boundaries = new ArrayList<>(boundariesSet); |
| nsBundleSplitAlgorithm = NamespaceBundleSplitAlgorithm.SPECIFIED_POSITIONS_DIVIDE_FORCE_ALGO; |
| } |
| final AtomicInteger counter = new AtomicInteger(0); |
| var childBundles = data.splitServiceUnitToDestBroker().keySet().stream() |
| .map(child -> bundleFactory.getBundle( |
| bundle.getNamespaceObject().toString(), child)) |
| .collect(Collectors.toList()); |
| this.splitServiceUnitOnceAndRetry(namespaceService, bundleFactory, nsBundleSplitAlgorithm, |
| bundle, childBundles, boundaries, data, counter, startTime, completionFuture); |
| return completionFuture; |
| } |
| |
| |
| |
| @VisibleForTesting |
| protected void splitServiceUnitOnceAndRetry(NamespaceService namespaceService, |
| NamespaceBundleFactory bundleFactory, |
| NamespaceBundleSplitAlgorithm algorithm, |
| NamespaceBundle parentBundle, |
| List<NamespaceBundle> childBundles, |
| List<Long> boundaries, |
| ServiceUnitStateData parentData, |
| AtomicInteger counter, |
| long startTime, |
| CompletableFuture<Void> completionFuture) { |
| ownChildBundles(childBundles, parentData) |
| .thenCompose(__ -> getSplitNamespaceBundles( |
| namespaceService, bundleFactory, algorithm, parentBundle, childBundles, boundaries)) |
| .thenCompose(namespaceBundles -> updateSplitNamespaceBundlesAsync( |
| namespaceService, bundleFactory, parentBundle, namespaceBundles)) |
| .thenAccept(__ -> // Update bundled_topic cache for load-report-generation |
| pulsar.getBrokerService().refreshTopicToStatsMaps(parentBundle)) |
| .thenAccept(__ -> pubAsync(parentBundle.toString(), new ServiceUnitStateData( |
| Deleted, null, parentData.sourceBroker(), getNextVersionId(parentData)))) |
| .thenAccept(__ -> { |
| double splitBundleTime = TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - startTime)); |
| log.info("Successfully split {} parent namespace-bundle to {} in {} ms", |
| parentBundle, childBundles, splitBundleTime); |
| namespaceService.onNamespaceBundleSplit(parentBundle); |
| completionFuture.complete(null); |
| }) |
| .exceptionally(ex -> { |
| // Retry several times on BadVersion |
| Throwable throwable = FutureUtil.unwrapCompletionException(ex); |
| if ((throwable instanceof MetadataStoreException.BadVersionException) |
| && (counter.incrementAndGet() < NamespaceService.BUNDLE_SPLIT_RETRY_LIMIT)) { |
| log.warn("Failed to update bundle range in metadata store. Retrying {} th / {} limit", |
| counter.get(), NamespaceService.BUNDLE_SPLIT_RETRY_LIMIT, ex); |
| pulsar.getExecutor().schedule(() -> splitServiceUnitOnceAndRetry( |
| namespaceService, bundleFactory, algorithm, parentBundle, childBundles, |
| boundaries, parentData, counter, startTime, completionFuture), |
| 100, MILLISECONDS); |
| } else { |
| // Retry enough, or meet other exception |
| String msg = format("Failed to split bundle %s, Retried %d th / %d limit, reason %s", |
| parentBundle.toString(), counter.get(), |
| NamespaceService.BUNDLE_SPLIT_RETRY_LIMIT, throwable.getMessage()); |
| log.warn(msg, throwable); |
| completionFuture.completeExceptionally( |
| new BrokerServiceException.ServiceUnitNotReadyException(msg)); |
| } |
| return null; |
| }); |
| } |
| |
| private CompletableFuture<Void> ownChildBundles(List<NamespaceBundle> childBundles, |
| ServiceUnitStateData parentData) { |
| List<CompletableFuture<Void>> futures = new ArrayList<>(childBundles.size()); |
| var debug = debug(); |
| for (var childBundle : childBundles) { |
| var childBundleStr = childBundle.toString(); |
| var childData = tableview.get(childBundleStr); |
| if (childData != null) { |
| if (debug) { |
| log.info("Already owned child bundle:{}", childBundleStr); |
| } |
| } else { |
| childData = new ServiceUnitStateData(Owned, parentData.sourceBroker(), |
| VERSION_ID_INIT); |
| futures.add(pubAsync(childBundleStr, childData).thenApply(__ -> null)); |
| } |
| } |
| |
| if (!futures.isEmpty()) { |
| return FutureUtil.waitForAll(futures); |
| } else { |
| return CompletableFuture.completedFuture(null); |
| } |
| } |
| |
| private CompletableFuture<NamespaceBundles> getSplitNamespaceBundles(NamespaceService namespaceService, |
| NamespaceBundleFactory bundleFactory, |
| NamespaceBundleSplitAlgorithm algorithm, |
| NamespaceBundle parentBundle, |
| List<NamespaceBundle> childBundles, |
| List<Long> boundaries) { |
| CompletableFuture future = new CompletableFuture(); |
| final var debug = debug(); |
| var targetNsBundle = bundleFactory.getBundles(parentBundle.getNamespaceObject()); |
| boolean found = false; |
| try { |
| targetNsBundle.validateBundle(parentBundle); |
| } catch (IllegalArgumentException e) { |
| if (debug) { |
| log.info("Namespace bundles do not contain the parent bundle:{}", |
| parentBundle); |
| } |
| for (var childBundle : childBundles) { |
| try { |
| targetNsBundle.validateBundle(childBundle); |
| if (debug) { |
| log.info("Namespace bundles contain the child bundle:{}", |
| childBundle); |
| } |
| } catch (Exception ex) { |
| future.completeExceptionally( |
| new BrokerServiceException.ServiceUnitNotReadyException( |
| "Namespace bundles do not contain the child bundle:" + childBundle, e)); |
| return future; |
| } |
| } |
| found = true; |
| } catch (Exception e) { |
| future.completeExceptionally( |
| new BrokerServiceException.ServiceUnitNotReadyException( |
| "Failed to validate the parent bundle in the namespace bundles.", e)); |
| return future; |
| } |
| if (found) { |
| future.complete(targetNsBundle); |
| return future; |
| } else { |
| return namespaceService.getSplitBoundary(parentBundle, algorithm, boundaries) |
| .thenApply(splitBundlesPair -> splitBundlesPair.getLeft()); |
| } |
| } |
| |
| private CompletableFuture<Void> updateSplitNamespaceBundlesAsync( |
| NamespaceService namespaceService, |
| NamespaceBundleFactory bundleFactory, |
| NamespaceBundle parentBundle, |
| NamespaceBundles splitNamespaceBundles) { |
| var namespaceName = parentBundle.getNamespaceObject(); |
| return namespaceService.updateNamespaceBundles( |
| namespaceName, splitNamespaceBundles) |
| .thenCompose(__ -> namespaceService.updateNamespaceBundlesForPolicies( |
| namespaceName, splitNamespaceBundles)) |
| .thenAccept(__ -> { |
| bundleFactory.invalidateBundleCache(parentBundle.getNamespaceObject()); |
| if (debug()) { |
| log.info("Successfully updated split namespace bundles and namespace bundle cache."); |
| } |
| }); |
| } |
| |
| public void handleMetadataSessionEvent(SessionEvent e) { |
| if (e == SessionReestablished || e == SessionLost) { |
| lastMetadataSessionEvent = e; |
| lastMetadataSessionEventTimestamp = System.currentTimeMillis(); |
| log.info("Received metadata session event:{} at timestamp:{}", |
| lastMetadataSessionEvent, lastMetadataSessionEventTimestamp); |
| } |
| } |
| |
| public void handleBrokerRegistrationEvent(String broker, NotificationType type) { |
| if (type == NotificationType.Created) { |
| log.info("BrokerRegistry detected the broker:{} registry has been created.", broker); |
| handleBrokerCreationEvent(broker); |
| } else if (type == NotificationType.Deleted) { |
| log.info("BrokerRegistry detected the broker:{} registry has been deleted.", broker); |
| handleBrokerDeletionEvent(broker); |
| } |
| } |
| |
| private MetadataState getMetadataState() { |
| long now = System.currentTimeMillis(); |
| if (lastMetadataSessionEvent == SessionReestablished) { |
| if (now - lastMetadataSessionEventTimestamp > 1000 * maxCleanupDelayTimeInSecs) { |
| return Stable; |
| } |
| return Jittery; |
| } |
| return Unstable; |
| } |
| |
| private void handleBrokerCreationEvent(String broker) { |
| CompletableFuture<Void> future = cleanupJobs.remove(broker); |
| if (future != null) { |
| future.cancel(false); |
| totalInactiveBrokerCleanupCancelledCnt++; |
| log.info("Successfully cancelled the ownership cleanup for broker:{}." |
| + " Active cleanup job count:{}", |
| broker, cleanupJobs.size()); |
| } else { |
| if (debug()) { |
| log.info("No needs to cancel the ownership cleanup for broker:{}." |
| + " There was no scheduled cleanup job. Active cleanup job count:{}", |
| broker, cleanupJobs.size()); |
| } |
| } |
| } |
| |
| private void handleBrokerDeletionEvent(String broker) { |
| if (!isChannelOwner()) { |
| log.warn("This broker is not the leader now. Ignoring BrokerDeletionEvent for broker {}.", broker); |
| return; |
| } |
| MetadataState state = getMetadataState(); |
| log.info("Handling broker:{} ownership cleanup based on metadata connection state:{}, event:{}, event_ts:{}:", |
| broker, state, lastMetadataSessionEvent, lastMetadataSessionEventTimestamp); |
| switch (state) { |
| case Stable -> scheduleCleanup(broker, minCleanupDelayTimeInSecs); |
| case Jittery -> scheduleCleanup(broker, maxCleanupDelayTimeInSecs); |
| case Unstable -> { |
| totalInactiveBrokerCleanupIgnoredCnt++; |
| log.error("MetadataState state is unstable. " |
| + "Ignoring the ownership cleanup request for the reported broker :{}", broker); |
| } |
| } |
| } |
| |
| private void scheduleCleanup(String broker, long delayInSecs) { |
| var scheduled = new MutableObject<CompletableFuture<Void>>(); |
| try { |
| cleanupJobs.computeIfAbsent(broker, k -> { |
| Executor delayed = CompletableFuture |
| .delayedExecutor(delayInSecs, TimeUnit.SECONDS, pulsar.getLoadManagerExecutor()); |
| totalInactiveBrokerCleanupScheduledCnt++; |
| var future = CompletableFuture |
| .runAsync(() -> { |
| try { |
| doCleanup(broker); |
| } catch (Throwable e) { |
| log.error("Failed to run the cleanup job for the broker {}, " |
| + "totalCleanupErrorCnt:{}.", |
| broker, totalCleanupErrorCnt.incrementAndGet(), e); |
| } |
| } |
| , delayed); |
| scheduled.setValue(future); |
| return future; |
| }); |
| } finally { |
| var future = scheduled.getValue(); |
| if (future != null) { |
| future.whenComplete((v, ex) -> { |
| cleanupJobs.remove(broker); |
| }); |
| } |
| } |
| |
| log.info("Scheduled ownership cleanup for broker:{} with delay:{} secs. Pending clean jobs:{}.", |
| broker, delayInSecs, cleanupJobs.size()); |
| } |
| |
| |
| private ServiceUnitStateData getOverrideInactiveBrokerStateData(ServiceUnitStateData orphanData, |
| String selectedBroker, |
| String inactiveBroker) { |
| if (orphanData.state() == Splitting) { |
| return new ServiceUnitStateData(Splitting, orphanData.dstBroker(), selectedBroker, |
| Map.copyOf(orphanData.splitServiceUnitToDestBroker()), |
| true, getNextVersionId(orphanData)); |
| } else { |
| return new ServiceUnitStateData(Owned, selectedBroker, inactiveBroker, |
| true, getNextVersionId(orphanData)); |
| } |
| } |
| |
| private void overrideOwnership(String serviceUnit, ServiceUnitStateData orphanData, String inactiveBroker) { |
| Optional<String> selectedBroker = selectBroker(serviceUnit, inactiveBroker); |
| if (selectedBroker.isPresent()) { |
| var override = getOverrideInactiveBrokerStateData( |
| orphanData, selectedBroker.get(), inactiveBroker); |
| log.info("Overriding ownership serviceUnit:{} from orphanData:{} to overrideData:{}", |
| serviceUnit, orphanData, override); |
| publishOverrideEventAsync(serviceUnit, orphanData, override) |
| .exceptionally(e -> { |
| log.error( |
| "Failed to override the ownership serviceUnit:{} orphanData:{}. " |
| + "Failed to publish override event. totalCleanupErrorCnt:{}", |
| serviceUnit, orphanData, totalCleanupErrorCnt.incrementAndGet()); |
| return null; |
| }); |
| } else { |
| log.error("Failed to override the ownership serviceUnit:{} orphanData:{}. Empty selected broker. " |
| + "totalCleanupErrorCnt:{}", |
| serviceUnit, orphanData, totalCleanupErrorCnt.incrementAndGet()); |
| } |
| } |
| |
| private void waitForCleanups(String broker, boolean excludeSystemTopics, int maxWaitTimeInMillis) { |
| long started = System.currentTimeMillis(); |
| while (System.currentTimeMillis() - started < maxWaitTimeInMillis) { |
| boolean cleaned = true; |
| for (var etr : tableview.entrySet()) { |
| var serviceUnit = etr.getKey(); |
| var data = etr.getValue(); |
| |
| if (excludeSystemTopics && serviceUnit.startsWith(SYSTEM_NAMESPACE.toString())) { |
| continue; |
| } |
| |
| if (data.state() == Owned && broker.equals(data.dstBroker())) { |
| cleaned = false; |
| break; |
| } |
| } |
| if (cleaned) { |
| try { |
| MILLISECONDS.sleep(OWNERSHIP_CLEAN_UP_CONVERGENCE_DELAY_IN_MILLIS); |
| } catch (InterruptedException e) { |
| log.warn("Interrupted while gracefully waiting for the cleanup convergence."); |
| } |
| break; |
| } else { |
| try { |
| MILLISECONDS.sleep(OWNERSHIP_CLEAN_UP_WAIT_RETRY_DELAY_IN_MILLIS); |
| } catch (InterruptedException e) { |
| log.warn("Interrupted while delaying the next service unit clean-up. Cleaning broker:{}", |
| lookupServiceAddress); |
| } |
| } |
| } |
| } |
| |
| private synchronized void doCleanup(String broker) { |
| long startTime = System.nanoTime(); |
| log.info("Started ownership cleanup for the inactive broker:{}", broker); |
| int orphanServiceUnitCleanupCnt = 0; |
| long totalCleanupErrorCntStart = totalCleanupErrorCnt.get(); |
| Map<String, ServiceUnitStateData> orphanSystemServiceUnits = new HashMap<>(); |
| for (var etr : tableview.entrySet()) { |
| var stateData = etr.getValue(); |
| var serviceUnit = etr.getKey(); |
| var state = state(stateData); |
| if (StringUtils.equals(broker, stateData.dstBroker()) && isActiveState(state) |
| || StringUtils.equals(broker, stateData.sourceBroker()) && isInFlightState(state)) { |
| if (serviceUnit.startsWith(SYSTEM_NAMESPACE.toString())) { |
| orphanSystemServiceUnits.put(serviceUnit, stateData); |
| } else { |
| overrideOwnership(serviceUnit, stateData, broker); |
| } |
| orphanServiceUnitCleanupCnt++; |
| } |
| } |
| |
| try { |
| producer.flush(); |
| } catch (PulsarClientException e) { |
| log.error("Failed to flush the in-flight non-system bundle override messages.", e); |
| } |
| |
| |
| if (orphanServiceUnitCleanupCnt > 0) { |
| // System bundles can contain this channel's system topic and other important system topics. |
| // Cleaning such system bundles(closing the system topics) together with the non-system bundles |
| // can cause the cluster to be temporarily unstable. |
| // Hence, we clean the non-system bundles first and gracefully wait for them. |
| // After that, we clean the system bundles, if any. |
| waitForCleanups(broker, true, OWNERSHIP_CLEAN_UP_MAX_WAIT_TIME_IN_MILLIS); |
| this.totalOrphanServiceUnitCleanupCnt += orphanServiceUnitCleanupCnt; |
| this.totalInactiveBrokerCleanupCnt++; |
| } |
| |
| // clean system bundles in the end |
| for (var orphanSystemServiceUnit : orphanSystemServiceUnits.entrySet()) { |
| log.info("Overriding orphan system service unit:{}", orphanSystemServiceUnit.getKey()); |
| overrideOwnership(orphanSystemServiceUnit.getKey(), orphanSystemServiceUnit.getValue(), broker); |
| } |
| |
| try { |
| producer.flush(); |
| } catch (PulsarClientException e) { |
| log.error("Failed to flush the in-flight system bundle override messages.", e); |
| } |
| |
| double cleanupTime = TimeUnit.NANOSECONDS |
| .toMillis((System.nanoTime() - startTime)); |
| |
| // clean load data stores |
| getContext().topBundleLoadDataStore().removeAsync(broker); |
| getContext().brokerLoadDataStore().removeAsync(broker); |
| |
| log.info("Completed a cleanup for the inactive broker:{} in {} ms. " |
| + "Cleaned up orphan service units: orphanServiceUnitCleanupCnt:{}, " |
| + "approximate cleanupErrorCnt:{}, metrics:{} ", |
| broker, |
| cleanupTime, |
| orphanServiceUnitCleanupCnt, |
| totalCleanupErrorCntStart - totalCleanupErrorCnt.get(), |
| printCleanupMetrics()); |
| |
| } |
| |
| private Optional<String> selectBroker(String serviceUnit, String inactiveBroker) { |
| try { |
| return loadManager.selectAsync( |
| LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit), Set.of(inactiveBroker)) |
| .get(inFlightStateWaitingTimeInMillis, MILLISECONDS); |
| } catch (Throwable e) { |
| log.error("Failed to select a broker for serviceUnit:{}", serviceUnit); |
| } |
| return Optional.empty(); |
| } |
| |
| private Optional<ServiceUnitStateData> getRollForwardStateData(String serviceUnit, |
| String inactiveBroker, |
| long nextVersionId) { |
| Optional<String> selectedBroker = selectBroker(serviceUnit, inactiveBroker); |
| if (selectedBroker.isEmpty()) { |
| return Optional.empty(); |
| } |
| return Optional.of(new ServiceUnitStateData(Owned, selectedBroker.get(), true, nextVersionId)); |
| } |
| |
| |
| private Optional<ServiceUnitStateData> getOverrideInFlightStateData( |
| String serviceUnit, ServiceUnitStateData orphanData, |
| Set<String> availableBrokers) { |
| long nextVersionId = getNextVersionId(orphanData); |
| var state = orphanData.state(); |
| switch (state) { |
| case Assigning: { |
| return getRollForwardStateData(serviceUnit, orphanData.dstBroker(), nextVersionId); |
| } |
| case Splitting: { |
| return Optional.of(new ServiceUnitStateData(Splitting, |
| orphanData.dstBroker(), orphanData.sourceBroker(), |
| Map.copyOf(orphanData.splitServiceUnitToDestBroker()), |
| true, nextVersionId)); |
| } |
| case Releasing: { |
| if (availableBrokers.contains(orphanData.sourceBroker())) { |
| // rollback to the src |
| return Optional.of(new ServiceUnitStateData(Owned, orphanData.sourceBroker(), true, nextVersionId)); |
| } else { |
| return getRollForwardStateData(serviceUnit, orphanData.sourceBroker(), nextVersionId); |
| } |
| } |
| default: { |
| var msg = String.format("Failed to get the overrideStateData from serviceUnit=%s, orphanData=%s", |
| serviceUnit, orphanData); |
| log.error(msg); |
| throw new IllegalStateException(msg); |
| } |
| } |
| } |
| |
| @VisibleForTesting |
| protected void monitorOwnerships(List<String> brokers) { |
| if (!isChannelOwner()) { |
| log.warn("This broker is not the leader now. Skipping ownership monitor."); |
| return; |
| } |
| |
| if (brokers == null || brokers.size() == 0) { |
| log.error("no active brokers found. Skipping ownership monitor."); |
| return; |
| } |
| |
| var metadataState = getMetadataState(); |
| if (metadataState != Stable) { |
| log.warn("metadata state:{} is not Stable. Skipping ownership monitor.", metadataState); |
| return; |
| } |
| |
| var debug = debug(); |
| if (debug) { |
| log.info("Started the ownership monitor run for activeBrokerCount:{}", brokers.size()); |
| } |
| |
| long startTime = System.nanoTime(); |
| Set<String> inactiveBrokers = new HashSet<>(); |
| Set<String> activeBrokers = new HashSet<>(brokers); |
| Map<String, ServiceUnitStateData> orphanServiceUnits = new HashMap<>(); |
| int serviceUnitTombstoneCleanupCnt = 0; |
| int orphanServiceUnitCleanupCnt = 0; |
| long totalCleanupErrorCntStart = totalCleanupErrorCnt.get(); |
| long now = System.currentTimeMillis(); |
| for (var etr : tableview.entrySet()) { |
| String serviceUnit = etr.getKey(); |
| ServiceUnitStateData stateData = etr.getValue(); |
| String dstBroker = stateData.dstBroker(); |
| String srcBroker = stateData.sourceBroker(); |
| var state = stateData.state(); |
| |
| if (isActiveState(state) && StringUtils.isNotBlank(srcBroker) && !activeBrokers.contains(srcBroker)) { |
| inactiveBrokers.add(srcBroker); |
| continue; |
| } |
| if (isActiveState(state) && StringUtils.isNotBlank(dstBroker) && !activeBrokers.contains(dstBroker)) { |
| inactiveBrokers.add(dstBroker); |
| continue; |
| } |
| if (isActiveState(state) && isInFlightState(state) |
| && now - stateData.timestamp() > inFlightStateWaitingTimeInMillis) { |
| orphanServiceUnits.put(serviceUnit, stateData); |
| continue; |
| } |
| |
| if (now - stateData.timestamp() > semiTerminalStateWaitingTimeInMillis) { |
| log.info("Found semi-terminal states to tombstone" |
| + " serviceUnit:{}, stateData:{}", serviceUnit, stateData); |
| tombstoneAsync(serviceUnit).whenComplete((__, e) -> { |
| if (e != null) { |
| log.error("Failed cleaning the ownership serviceUnit:{}, stateData:{}, " |
| + "cleanupErrorCnt:{}.", |
| serviceUnit, stateData, |
| totalCleanupErrorCnt.incrementAndGet() - totalCleanupErrorCntStart, e); |
| } |
| }); |
| serviceUnitTombstoneCleanupCnt++; |
| } |
| } |
| |
| // Skip cleaning orphan bundles if inactiveBrokers exist. This is a bigger problem. |
| if (!inactiveBrokers.isEmpty()) { |
| for (String inactiveBroker : inactiveBrokers) { |
| handleBrokerDeletionEvent(inactiveBroker); |
| } |
| } else if (!orphanServiceUnits.isEmpty()) { |
| for (var etr : orphanServiceUnits.entrySet()) { |
| var orphanServiceUnit = etr.getKey(); |
| var orphanData = etr.getValue(); |
| var overrideData = getOverrideInFlightStateData( |
| orphanServiceUnit, orphanData, activeBrokers); |
| if (overrideData.isPresent()) { |
| log.info("Overriding in-flight state ownership serviceUnit:{} " |
| + "from orphanData:{} to overrideData:{}", |
| orphanServiceUnit, orphanData, overrideData); |
| publishOverrideEventAsync(orphanServiceUnit, orphanData, overrideData.get()) |
| .whenComplete((__, e) -> { |
| if (e != null) { |
| log.error("Failed cleaning the ownership orphanServiceUnit:{}, orphanData:{}, " |
| + "cleanupErrorCnt:{}.", |
| orphanServiceUnit, orphanData, |
| totalCleanupErrorCnt.incrementAndGet() - totalCleanupErrorCntStart, e); |
| } |
| }); |
| orphanServiceUnitCleanupCnt++; |
| } else { |
| log.warn("Failed get the overrideStateData from orphanServiceUnit:{}, orphanData:{}," |
| + " cleanupErrorCnt:{}. will retry..", |
| orphanServiceUnit, orphanData, |
| totalCleanupErrorCnt.incrementAndGet() - totalCleanupErrorCntStart); |
| } |
| } |
| } |
| |
| try { |
| producer.flush(); |
| } catch (PulsarClientException e) { |
| log.error("Failed to flush the in-flight messages.", e); |
| } |
| |
| boolean cleaned = false; |
| if (serviceUnitTombstoneCleanupCnt > 0) { |
| this.totalServiceUnitTombstoneCleanupCnt += serviceUnitTombstoneCleanupCnt; |
| cleaned = true; |
| } |
| |
| if (orphanServiceUnitCleanupCnt > 0) { |
| this.totalOrphanServiceUnitCleanupCnt += orphanServiceUnitCleanupCnt; |
| cleaned = true; |
| } |
| |
| if (debug || cleaned) { |
| double monitorTime = TimeUnit.NANOSECONDS |
| .toMillis((System.nanoTime() - startTime)); |
| log.info("Completed the ownership monitor run in {} ms. " |
| + "Scheduled cleanups for inactive brokers:{}. inactiveBrokerCount:{}. " |
| + "Published cleanups for orphan service units, orphanServiceUnitCleanupCnt:{}. " |
| + "Tombstoned semi-terminal state service units, serviceUnitTombstoneCleanupCnt:{}. " |
| + "Approximate cleanupErrorCnt:{}, metrics:{}. ", |
| monitorTime, |
| inactiveBrokers, inactiveBrokers.size(), |
| orphanServiceUnitCleanupCnt, |
| serviceUnitTombstoneCleanupCnt, |
| totalCleanupErrorCntStart - totalCleanupErrorCnt.get(), |
| printCleanupMetrics()); |
| } |
| |
| } |
| |
| private String printCleanupMetrics() { |
| return String.format( |
| "{totalInactiveBrokerCleanupCnt:%d, " |
| + "totalServiceUnitTombstoneCleanupCnt:%d, totalOrphanServiceUnitCleanupCnt:%d, " |
| + "totalCleanupErrorCnt:%d, " |
| + "totalInactiveBrokerCleanupScheduledCnt%d, totalInactiveBrokerCleanupIgnoredCnt:%d, " |
| + "totalInactiveBrokerCleanupCancelledCnt:%d, " |
| + " activeCleanupJobs:%d}", |
| totalInactiveBrokerCleanupCnt, |
| totalServiceUnitTombstoneCleanupCnt, |
| totalOrphanServiceUnitCleanupCnt, |
| totalCleanupErrorCnt.get(), |
| totalInactiveBrokerCleanupScheduledCnt, |
| totalInactiveBrokerCleanupIgnoredCnt, |
| totalInactiveBrokerCleanupCancelledCnt, |
| cleanupJobs.size() |
| ); |
| } |
| |
| private int getTotalOwnedServiceUnitCnt() { |
| if (tableview == null) { |
| return 0; |
| } |
| long now = System.currentTimeMillis(); |
| if (lastOwnEventHandledAt > lastOwnedServiceUnitCountAt |
| || now - lastOwnedServiceUnitCountAt > MAX_OWNED_BUNDLE_COUNT_DELAY_TIME_IN_MILLIS) { |
| int cnt = 0; |
| for (var data : tableview.values()) { |
| if (data.state() == Owned && isTargetBroker(data.dstBroker())) { |
| cnt++; |
| } |
| } |
| lastOwnedServiceUnitCountAt = now; |
| totalOwnedServiceUnitCnt = cnt; |
| } |
| return totalOwnedServiceUnitCnt; |
| } |
| |
| |
| @Override |
| public List<Metrics> getMetrics() { |
| var metrics = new ArrayList<Metrics>(); |
| var dimensions = new HashMap<String, String>(); |
| dimensions.put("metric", "sunitStateChn"); |
| dimensions.put("broker", pulsar.getAdvertisedAddress()); |
| |
| for (var etr : ownerLookUpCounters.entrySet()) { |
| { |
| var dim = new HashMap<>(dimensions); |
| dim.put("state", etr.getKey().toString()); |
| dim.put("result", "Total"); |
| var metric = Metrics.create(dim); |
| metric.put("brk_sunit_state_chn_owner_lookup_total", |
| etr.getValue().getTotal().get()); |
| metrics.add(metric); |
| } |
| |
| { |
| var dim = new HashMap<>(dimensions); |
| dim.put("state", etr.getKey().toString()); |
| dim.put("result", "Failure"); |
| var metric = Metrics.create(dim); |
| metric.put("brk_sunit_state_chn_owner_lookup_total", |
| etr.getValue().getFailure().get()); |
| metrics.add(metric); |
| } |
| } |
| |
| for (var etr : eventCounters.entrySet()) { |
| { |
| var dim = new HashMap<>(dimensions); |
| dim.put("event", etr.getKey().toString()); |
| dim.put("result", "Total"); |
| var metric = Metrics.create(dim); |
| metric.put("brk_sunit_state_chn_event_publish_ops_total", |
| etr.getValue().getTotal().get()); |
| metrics.add(metric); |
| } |
| |
| { |
| var dim = new HashMap<>(dimensions); |
| dim.put("event", etr.getKey().toString()); |
| dim.put("result", "Failure"); |
| var metric = Metrics.create(dim); |
| metric.put("brk_sunit_state_chn_event_publish_ops_total", |
| etr.getValue().getFailure().get()); |
| metrics.add(metric); |
| } |
| } |
| |
| for (var etr : handlerCounters.entrySet()) { |
| { |
| var dim = new HashMap<>(dimensions); |
| dim.put("event", etr.getKey().toString()); |
| dim.put("result", "Total"); |
| var metric = Metrics.create(dim); |
| metric.put("brk_sunit_state_chn_subscribe_ops_total", |
| etr.getValue().getTotal().get()); |
| metrics.add(metric); |
| } |
| |
| { |
| var dim = new HashMap<>(dimensions); |
| dim.put("event", etr.getKey().toString()); |
| dim.put("result", "Failure"); |
| var metric = Metrics.create(dim); |
| metric.put("brk_sunit_state_chn_subscribe_ops_total", |
| etr.getValue().getFailure().get()); |
| metrics.add(metric); |
| } |
| } |
| |
| { |
| var dim = new HashMap<>(dimensions); |
| dim.put("result", "Failure"); |
| var metric = Metrics.create(dim); |
| metric.put("brk_sunit_state_chn_cleanup_ops_total", totalCleanupErrorCnt.get()); |
| metrics.add(metric); |
| } |
| |
| { |
| var dim = new HashMap<>(dimensions); |
| dim.put("result", "Skip"); |
| var metric = Metrics.create(dim); |
| metric.put("brk_sunit_state_chn_inactive_broker_cleanup_ops_total", totalInactiveBrokerCleanupIgnoredCnt); |
| metrics.add(metric); |
| } |
| |
| { |
| var dim = new HashMap<>(dimensions); |
| dim.put("result", "Cancel"); |
| var metric = Metrics.create(dim); |
| metric.put("brk_sunit_state_chn_inactive_broker_cleanup_ops_total", totalInactiveBrokerCleanupCancelledCnt); |
| metrics.add(metric); |
| } |
| |
| { |
| var dim = new HashMap<>(dimensions); |
| dim.put("result", "Schedule"); |
| var metric = Metrics.create(dim); |
| metric.put("brk_sunit_state_chn_inactive_broker_cleanup_ops_total", totalInactiveBrokerCleanupScheduledCnt); |
| metrics.add(metric); |
| } |
| |
| { |
| var dim = new HashMap<>(dimensions); |
| dim.put("result", "Success"); |
| var metric = Metrics.create(dim); |
| metric.put("brk_sunit_state_chn_inactive_broker_cleanup_ops_total", totalInactiveBrokerCleanupCnt); |
| metrics.add(metric); |
| } |
| |
| var metric = Metrics.create(dimensions); |
| |
| metric.put("brk_sunit_state_chn_orphan_su_cleanup_ops_total", totalOrphanServiceUnitCleanupCnt); |
| metric.put("brk_sunit_state_chn_su_tombstone_cleanup_ops_total", totalServiceUnitTombstoneCleanupCnt); |
| metric.put("brk_sunit_state_chn_owned_su_total", getTotalOwnedServiceUnitCnt()); |
| metrics.add(metric); |
| |
| return metrics; |
| } |
| |
| @Override |
| public void listen(StateChangeListener listener) { |
| this.stateChangeListeners.addListener(listener); |
| |
| } |
| |
| @Override |
| public Set<Map.Entry<String, ServiceUnitStateData>> getOwnershipEntrySet() { |
| return tableview.entrySet(); |
| } |
| |
| public static ServiceUnitStateChannel get(PulsarService pulsar) { |
| return ExtensibleLoadManagerImpl.get(pulsar.getLoadManager().get()).getServiceUnitStateChannel(); |
| } |
| } |
