Skip to content

PIP-45: Implement load-manager leader election using new CoordinationService interface #9240

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merlimat merged 10 commits into from
Jan 22, 2021
Merged

PIP-45: Implement load-manager leader election using new CoordinationService interface #9240

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
97f7de9
PIP-45: Implement load-manager leader election using new Coordination…
merlimat Jan 19, 2021
0afb503
Fixed checkstyle issue
merlimat Jan 19, 2021
affdb4a
Allow to mock the metadata instance creation
merlimat Jan 20, 2021
e4f56ef
More test fixes
merlimat Jan 20, 2021
4256167
Fixed style
merlimat Jan 20, 2021
8361085
Test fixes
merlimat Jan 20, 2021
66080b8
Cancel task if there's one already there.
merlimat Jan 20, 2021
3fef1e6
Merge remote-tracking branch 'apache/master' into load-manager-leader…
merlimat Jan 20, 2021
69ef553
Test fixes
merlimat Jan 20, 2021
118417c
Merge remote-tracking branch 'apache/master' into load-manager-leader…
merlimat Jan 21, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 84 additions & 33 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
Expand Down Expand Up @@ -74,7 +75,6 @@
import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.intercept.BrokerInterceptors;
import org.apache.pulsar.broker.loadbalance.LeaderElectionService;
import org.apache.pulsar.broker.loadbalance.LeaderElectionService.LeaderListener;
import org.apache.pulsar.broker.loadbalance.LoadManager;
import org.apache.pulsar.broker.loadbalance.LoadReportUpdaterTask;
import org.apache.pulsar.broker.loadbalance.LoadResourceQuotaUpdaterTask;
Expand Down Expand Up @@ -115,6 +115,12 @@
import org.apache.pulsar.functions.worker.ErrorNotifier;
import org.apache.pulsar.functions.worker.WorkerConfig;
import org.apache.pulsar.functions.worker.WorkerService;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.coordination.CoordinationService;
import org.apache.pulsar.metadata.api.coordination.LeaderElectionState;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
import org.apache.pulsar.metadata.coordination.impl.CoordinationServiceImpl;
import org.apache.pulsar.packages.management.core.PackagesManagement;
import org.apache.pulsar.packages.management.core.PackagesStorage;
import org.apache.pulsar.packages.management.core.PackagesStorageProvider;
Expand Down Expand Up @@ -209,6 +215,9 @@ public class PulsarService implements AutoCloseable {
private PrometheusMetricsServlet metricsServlet;
private List<PrometheusRawMetricsProvider> pendingMetricsProviders;

private MetadataStoreExtended localMetadataStore;
private CoordinationService coordinationService;


public enum State {
Init, Started, Closed
Expand Down Expand Up @@ -267,7 +276,7 @@ public PulsarService(ServiceConfiguration config,
new DefaultThreadFactory("pulsar"));
this.cacheExecutor = Executors.newScheduledThreadPool(config.getNumCacheExecutorThreadPoolSize(),
new DefaultThreadFactory("zk-cache-callback"));
}
}

/**
* Close the current pulsar service. All resources are released.
Expand Down Expand Up @@ -307,7 +316,7 @@ public void close() throws PulsarServerException {
}

if (this.leaderElectionService != null) {
this.leaderElectionService.stop();
this.leaderElectionService.close();
this.leaderElectionService = null;
}

Expand Down Expand Up @@ -378,10 +387,25 @@ public void close() throws PulsarServerException {
transactionBufferClient.close();
}

if (coordinationService != null) {
coordinationService.close();
}

if (localMetadataStore != null) {
localMetadataStore.close();
}

state = State.Closed;
isClosedCondition.signalAll();
} catch (Exception e) {
throw new PulsarServerException(e);
if (e instanceof CompletionException && e.getCause() instanceof MetadataStoreException) {
throw new PulsarServerException(MetadataStoreException.unwrap((CompletionException) e));
} else if (e.getCause() instanceof CompletionException
&& e.getCause().getCause() instanceof MetadataStoreException) {
throw new PulsarServerException(MetadataStoreException.unwrap((CompletionException) e.getCause()));
} else {
throw new PulsarServerException(e);
}
} finally {
mutex.unlock();
}
Expand Down Expand Up @@ -440,6 +464,10 @@ public void start() throws PulsarServerException {
throw new IllegalArgumentException("brokerServicePort/brokerServicePortTls must be present");
}

localMetadataStore = createLocalMetadataStore();

coordinationService = new CoordinationServiceImpl(localMetadataStore);

orderedExecutor = OrderedExecutor.newBuilder()
.numThreads(config.getNumOrderedExecutorThreads())
.name("pulsar-ordered")
Expand Down Expand Up @@ -646,36 +674,51 @@ public Boolean get() {
}
}

protected void startLeaderElectionService() {
this.leaderElectionService = new LeaderElectionService(this, new LeaderListener() {
@Override
public synchronized void brokerIsTheLeaderNow() {
if (getConfiguration().isLoadBalancerEnabled()) {
long loadSheddingInterval = TimeUnit.MINUTES
.toMillis(getConfiguration().getLoadBalancerSheddingIntervalMinutes());
long resourceQuotaUpdateInterval = TimeUnit.MINUTES
.toMillis(getConfiguration().getLoadBalancerResourceQuotaUpdateIntervalMinutes());

loadSheddingTask = loadManagerExecutor.scheduleAtFixedRate(new LoadSheddingTask(loadManager),
loadSheddingInterval, loadSheddingInterval, TimeUnit.MILLISECONDS);
loadResourceQuotaTask = loadManagerExecutor.scheduleAtFixedRate(
new LoadResourceQuotaUpdaterTask(loadManager), resourceQuotaUpdateInterval,
resourceQuotaUpdateInterval, TimeUnit.MILLISECONDS);
}
}
public MetadataStoreExtended createLocalMetadataStore() throws MetadataStoreException {
return MetadataStoreExtended.create(config.getZookeeperServers(),
MetadataStoreConfig.builder()
.sessionTimeoutMillis((int) config.getZooKeeperSessionTimeoutMillis())
.allowReadOnlyOperations(false)
.build());
}

@Override
public synchronized void brokerIsAFollowerNow() {
if (loadSheddingTask != null) {
loadSheddingTask.cancel(false);
loadSheddingTask = null;
}
if (loadResourceQuotaTask != null) {
loadResourceQuotaTask.cancel(false);
loadResourceQuotaTask = null;
}
}
});
protected void startLeaderElectionService() {
this.leaderElectionService = new LeaderElectionService(coordinationService, getSafeWebServiceAddress(),
state -> {
if (state == LeaderElectionState.Leading) {
LOG.info("This broker was elected leader");
if (getConfiguration().isLoadBalancerEnabled()) {
long loadSheddingInterval = TimeUnit.MINUTES
.toMillis(getConfiguration().getLoadBalancerSheddingIntervalMinutes());
long resourceQuotaUpdateInterval = TimeUnit.MINUTES
.toMillis(getConfiguration().getLoadBalancerResourceQuotaUpdateIntervalMinutes());

if (loadSheddingTask != null) {
loadSheddingTask.cancel(false);
}
if (loadResourceQuotaTask != null) {
loadResourceQuotaTask.cancel(false);
}
loadSheddingTask = loadManagerExecutor.scheduleAtFixedRate(

@eolivelli eolivelli Jan 20, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

may it happen that we already have a value here in loadSheddingTask ? (some rare bug in the LeaderElectionService ?)
should be try to cancel it ?, the same for loadResourceQuotaTask

@merlimat merlimat Jan 20, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In theory it shouldn't happen, but it's a good point, I've added the cancel if it's already set.

new LoadSheddingTask(loadManager),
loadSheddingInterval, loadSheddingInterval, TimeUnit.MILLISECONDS);
loadResourceQuotaTask = loadManagerExecutor.scheduleAtFixedRate(
new LoadResourceQuotaUpdaterTask(loadManager), resourceQuotaUpdateInterval,
resourceQuotaUpdateInterval, TimeUnit.MILLISECONDS);
}
} else {
LOG.info("This broker is a follower. Current leader is {}",
leaderElectionService.getCurrentLeader());
if (loadSheddingTask != null) {
loadSheddingTask.cancel(false);
loadSheddingTask = null;
}
if (loadResourceQuotaTask != null) {
loadResourceQuotaTask.cancel(false);
loadResourceQuotaTask = null;
}
}
});

leaderElectionService.start();
}
Expand Down Expand Up @@ -1313,6 +1356,14 @@ public Optional<Integer> getBrokerListenPortTls() {
return brokerService.getListenPortTls();
}

public MetadataStoreExtended getLocalMetadataStore() {
return localMetadataStore;
}

public CoordinationService getCoordinationService() {
return coordinationService;
}

public static WorkerConfig initializeWorkerConfigFromBrokerConfig(ServiceConfiguration brokerConfig,
String workerConfigFile) throws IOException {
WorkerConfig workerConfig = WorkerConfig.load(workerConfigFile);
Expand Down
49 changes: 7 additions & 42 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderBroker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,52 +18,17 @@
*/
package org.apache.pulsar.broker.loadbalance;

import com.google.common.base.Objects;
import java.util.concurrent.atomic.AtomicBoolean;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
* A class to hold the contents of the leader election node. Facilitates serialization and deserialization of the
* information that might be added for leader broker in the future.
*
*
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class LeaderBroker {
public final String serviceUrl;

private AtomicBoolean isLeaderReady = new AtomicBoolean(false);

// Need this default constructor for json conversion. Please do not remove this.
public LeaderBroker() {
this(null);
}

public LeaderBroker(String serviceUrl) {
this.serviceUrl = serviceUrl;
}

public String getServiceUrl() {
return this.serviceUrl;
}

@Override
public int hashCode() {
return Objects.hashCode(serviceUrl);
}

@Override
public boolean equals(Object obj) {
if (obj instanceof LeaderBroker) {
LeaderBroker other = (LeaderBroker) obj;
return Objects.equal(serviceUrl, other.serviceUrl);
}
return false;
}

public boolean isLeaderReady() {
return isLeaderReady.get();
}

public void setLeaderReady(boolean isLeaderReady) {
this.isLeaderReady.compareAndSet(!isLeaderReady, isLeaderReady);
}
private String serviceUrl;
}
Loading