From 3772cf3c261c65e3a45fa6da7670aa27d96efb6d Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Fri, 3 Jul 2020 19:18:51 -0700 Subject: [PATCH 1/4] Allow function rebalance to be run periodically --- conf/functions_worker.yml | 2 ++ .../pulsar/functions/worker/WorkerConfig.java | 5 ++++ .../functions/worker/SchedulerManager.java | 23 ++++++++++++++++--- .../functions/worker/WorkerService.java | 13 +++++++++++ .../functions/worker/rest/api/WorkerImpl.java | 8 ++++--- 5 files changed, 45 insertions(+), 6 deletions(-) diff --git a/conf/functions_worker.yml b/conf/functions_worker.yml index 7ab160e12edf6..1f360eb0b4482 100644 --- a/conf/functions_worker.yml +++ b/conf/functions_worker.yml @@ -93,6 +93,8 @@ schedulerClassName: "org.apache.pulsar.functions.worker.scheduler.RoundRobinSche functionAssignmentTopicName: "assignments" failureCheckFreqMs: 30000 rescheduleTimeoutMs: 60000 +# frequency at which to check if cluster needs rebalancing (15min default interval) +rebalanceCheckFreqSec: 900 initialBrokerReconnectMaxRetries: 60 assignmentWriteMaxRetries: 60 instanceLivenessCheckFreqMs: 30000 diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java index 4b0ddd1f3af05..631ae6fe05092 100644 --- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java +++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java @@ -240,6 +240,11 @@ public class WorkerConfig implements Serializable, PulsarConfiguration { doc = "The reschedule timeout of function assignment, in milliseconds" ) private long rescheduleTimeoutMs; + @FieldContext( + category = CATEGORY_FUNC_RUNTIME_MNG, + doc = "The frequency to check whether the cluster needs rebalancing" + ) + private long rebalanceCheckFreqSec; @FieldContext( category = CATEGORY_FUNC_RUNTIME_MNG, doc = "The max number of retries for initial broker reconnects when function metadata manager" diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java index 43413102e77c1..ce4a91d7b501c 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java @@ -35,6 +35,7 @@ import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.common.util.Reflections; +import org.apache.pulsar.common.util.RestException; import org.apache.pulsar.functions.proto.Function; import org.apache.pulsar.functions.proto.Function.Assignment; import org.apache.pulsar.functions.proto.Function.FunctionDetails; @@ -44,6 +45,7 @@ import org.apache.pulsar.functions.utils.FunctionCommon; import org.apache.pulsar.functions.worker.scheduler.IScheduler; +import javax.ws.rs.core.Response; import java.util.HashMap; import java.util.Iterator; import java.util.LinkedList; @@ -113,6 +115,8 @@ public class SchedulerManager implements AutoCloseable { private MessageId lastMessageProduced = null; private MessageId metadataTopicLastMessage = MessageId.earliest; + private Future currentRebalanceFuture; + private AtomicBoolean rebalanceInProgess = new AtomicBoolean(false); public SchedulerManager(WorkerConfig workerConfig, PulsarClient pulsarClient, @@ -222,9 +226,18 @@ public Future schedule() { return scheduleInternal(() -> invokeScheduler(), "Encountered error when invoking scheduler"); } - public Future rebalance() { + private Future rebalance() { return scheduleInternal(() -> invokeRebalance(), "Encountered error when invoking rebalance"); } + + public Future rebalanceIfNotInprogress() { + if (rebalanceInProgess.compareAndSet(false, true)) { + currentRebalanceFuture = rebalance(); + return currentRebalanceFuture; + } else { + throw new RebalanceInProgressException(); + } + } @VisibleForTesting void invokeScheduler() { @@ -273,7 +286,7 @@ void invokeScheduler() { MessageId messageId = publishNewAssignment(newAssignment, false); // Directly update in memory assignment cache since I am leader - log.info("Updating assignment: {}", assignment); + log.info("Updating assignment: {}", newAssignment); functionRuntimeManager.processAssignment(newAssignment); // update message id associated with current view of assignments map lastMessageProduced = messageId; @@ -358,7 +371,8 @@ private void invokeRebalance() { // update message id associated with current view of assignments map lastMessageProduced = messageId; } - log.info("Total number of new assignments computed for rebalance: {}", rebalancedAssignments.size()); + log.info("Rebalance - Total number of new assignments computed: {}", rebalancedAssignments.size()); + rebalanceInProgess.set(false); } private void scheduleCompaction(ScheduledExecutorService executor, long scheduleFrequencySec) { @@ -511,4 +525,7 @@ static String checkHeartBeatFunction(Instance funInstance) { } return null; } + + public static class RebalanceInProgressException extends RuntimeException { + } } diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java index 7754a823219e1..9bf7fb3908eb9 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java @@ -39,6 +39,7 @@ import org.apache.pulsar.client.api.PulsarClientException; import java.net.URI; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -257,6 +258,18 @@ public void start(URI dlogUri, } }); + clusterServiceCoordinator.addTask("rebalance-periodic-check", + workerConfig.getRebalanceCheckFreqSec() * 1000, + () -> { + try { + schedulerManager.rebalanceIfNotInprogress().get(); + } catch (SchedulerManager.RebalanceInProgressException e) { + log.info("Scheduled for rebalance but rebalance is already in progress. Ignoring."); + } catch (Exception e) { + log.warn("Encountered error when running scheduled rebalance", e); + } + }); + log.info("/** Starting Cluster Service Coordinator **/"); clusterServiceCoordinator.start(); diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/WorkerImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/WorkerImpl.java index f58f57bda1435..17eab572d96bf 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/WorkerImpl.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/WorkerImpl.java @@ -29,6 +29,7 @@ import org.apache.pulsar.functions.worker.FunctionRuntimeInfo; import org.apache.pulsar.functions.worker.FunctionRuntimeManager; import org.apache.pulsar.functions.worker.MembershipManager; +import org.apache.pulsar.functions.worker.SchedulerManager; import org.apache.pulsar.functions.worker.WorkerService; import org.apache.pulsar.functions.worker.WorkerUtils; @@ -43,6 +44,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.function.Supplier; @@ -217,9 +219,9 @@ public void rebalance(final URI uri, final String clientRole) { } if (worker().getLeaderService().isLeader()) { - if (currentRebalanceFuture == null || currentRebalanceFuture.isDone()) { - currentRebalanceFuture = this.worker().getSchedulerManager().rebalance(); - } else { + try { + worker().getSchedulerManager().rebalanceIfNotInprogress(); + } catch (SchedulerManager.RebalanceInProgressException e) { throw new RestException(Status.BAD_REQUEST, "Rebalance already in progress"); } } else { From 246934f9724067a1ed9b828d8a28cfea03abd6da Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Fri, 3 Jul 2020 19:21:40 -0700 Subject: [PATCH 2/4] cleaning up --- .../org/apache/pulsar/functions/worker/SchedulerManager.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java index ce4a91d7b501c..374512acea01f 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java @@ -35,7 +35,6 @@ import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.common.util.Reflections; -import org.apache.pulsar.common.util.RestException; import org.apache.pulsar.functions.proto.Function; import org.apache.pulsar.functions.proto.Function.Assignment; import org.apache.pulsar.functions.proto.Function.FunctionDetails; @@ -45,7 +44,6 @@ import org.apache.pulsar.functions.utils.FunctionCommon; import org.apache.pulsar.functions.worker.scheduler.IScheduler; -import javax.ws.rs.core.Response; import java.util.HashMap; import java.util.Iterator; import java.util.LinkedList; From 12040fc4a20e8f77aca7ea5497d948abcb8b1158 Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Mon, 6 Jul 2020 11:18:49 -0700 Subject: [PATCH 3/4] improving --- conf/functions_worker.yml | 4 +-- .../functions/worker/WorkerService.java | 25 ++++++++++--------- .../functions/worker/rest/api/WorkerImpl.java | 1 - 3 files changed, 15 insertions(+), 15 deletions(-) diff --git a/conf/functions_worker.yml b/conf/functions_worker.yml index 1f360eb0b4482..d3715e0254451 100644 --- a/conf/functions_worker.yml +++ b/conf/functions_worker.yml @@ -93,8 +93,8 @@ schedulerClassName: "org.apache.pulsar.functions.worker.scheduler.RoundRobinSche functionAssignmentTopicName: "assignments" failureCheckFreqMs: 30000 rescheduleTimeoutMs: 60000 -# frequency at which to check if cluster needs rebalancing (15min default interval) -rebalanceCheckFreqSec: 900 +# frequency at which to check if cluster needs rebalancing (set to -1 to disable) +rebalanceCheckFreqSec: 3600 initialBrokerReconnectMaxRetries: 60 assignmentWriteMaxRetries: 60 instanceLivenessCheckFreqMs: 30000 diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java index 9bf7fb3908eb9..5b493bc09dbe5 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerService.java @@ -39,7 +39,6 @@ import org.apache.pulsar.client.api.PulsarClientException; import java.net.URI; -import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -258,17 +257,19 @@ public void start(URI dlogUri, } }); - clusterServiceCoordinator.addTask("rebalance-periodic-check", - workerConfig.getRebalanceCheckFreqSec() * 1000, - () -> { - try { - schedulerManager.rebalanceIfNotInprogress().get(); - } catch (SchedulerManager.RebalanceInProgressException e) { - log.info("Scheduled for rebalance but rebalance is already in progress. Ignoring."); - } catch (Exception e) { - log.warn("Encountered error when running scheduled rebalance", e); - } - }); + if (workerConfig.getRebalanceCheckFreqSec() > 0) { + clusterServiceCoordinator.addTask("rebalance-periodic-check", + workerConfig.getRebalanceCheckFreqSec() * 1000, + () -> { + try { + schedulerManager.rebalanceIfNotInprogress().get(); + } catch (SchedulerManager.RebalanceInProgressException e) { + log.info("Scheduled for rebalance but rebalance is already in progress. Ignoring."); + } catch (Exception e) { + log.warn("Encountered error when running scheduled rebalance", e); + } + }); + } log.info("/** Starting Cluster Service Coordinator **/"); clusterServiceCoordinator.start(); diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/WorkerImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/WorkerImpl.java index 17eab572d96bf..9e24f38862c32 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/WorkerImpl.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/WorkerImpl.java @@ -44,7 +44,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.function.Supplier; From 3b01578332d028f7a0c949a5c76030648f644b04 Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Mon, 6 Jul 2020 11:44:33 -0700 Subject: [PATCH 4/4] changing default --- conf/functions_worker.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/conf/functions_worker.yml b/conf/functions_worker.yml index d3715e0254451..1b0b9f89b393a 100644 --- a/conf/functions_worker.yml +++ b/conf/functions_worker.yml @@ -94,7 +94,7 @@ functionAssignmentTopicName: "assignments" failureCheckFreqMs: 30000 rescheduleTimeoutMs: 60000 # frequency at which to check if cluster needs rebalancing (set to -1 to disable) -rebalanceCheckFreqSec: 3600 +rebalanceCheckFreqSec: -1 initialBrokerReconnectMaxRetries: 60 assignmentWriteMaxRetries: 60 instanceLivenessCheckFreqMs: 30000