Fix leader/scheduler assignment processing lag problem #7237

jerrypeng · 2020-06-10T20:17:19Z

Motivation

When the leader worker isn't processing assignment messages fast enough. The background routine that checks for unassigned functions instances will trigger scheduler to schedule and write more assignments to the assignment topic. There is essentially a feedback loop that can cause many assignment updates to be published in the assignment topic that are unnecessary.

Modifications

Allow leader to modify/update locally in-memory assignments map

jerrypeng · 2020-06-10T22:27:05Z

/pulsarbot run-failure-checks

jerrypeng · 2020-06-11T00:24:39Z

/pulsarbot run-failure-checks

srkukarni · 2020-06-12T02:31:21Z

...ctions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java

            }
+
+            hasExited = new CompletableFuture<>();


this isn't right?

it is re-initializing the variable, so if we can "start" again, the completeable future is not already completed

Do you think its better to recreate the object? That way this re-create logic becomes simpler

Creating a new FunctionAssignmentTailer doesn't really simplify the logic much. "hasExited" is needed regardless of whether we recreate the object from scratch or not. We are also keeping the track of the "lastMessageId" in FunctionAssignmentTailer.

Then maybe we can create this at start instead?

I just dislike creating new objects in something like close. Seems like not the usual pattern

ok I re-initialize it in the start method

srkukarni · 2020-06-12T02:32:57Z

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/LeaderService.java

+            try {
+                // trigger read to the end of the topic and exit
+                // Since the leader can just update its in memory assignments cache directly
+                functionAssignmentTailer.triggerReadToTheEndAndExit().get();


This should be abstracted out from leaderservice to respective class(in this case functionruntimemanager)

Also we need to create the producer here right?

This should be abstracted out from leaderservice to respective class(in this case functionruntimemanager)

yup done

Also we need to create the producer here right?

Why do we need to create a producer? To start producing to the assignment topic? We initialize the producer in the constructor. I guess we don't need to do that and only when the worker becomes the leader will it create the producer and close the producer when it looses leadership

Yup. That is the same pattern in #7255 as well

srkukarni · 2020-06-12T02:35:49Z

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java

-                    } catch (Exception e) {
-                        log.warn("Failed to invoke scheduler", e);
-                        throw e;
+        try {


I think we need to simplify this massively.
I think part of the pr that I'm working on wrt metadata simplification will impact this as well.

What are you thinking? What is the complexity here?

srkukarni · 2020-06-14T20:53:56Z

...ctions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java

+                        lastMessageId = msg.getMessageId();
+                    }
+                } catch (Throwable th) {
+                    if (isRunning) {


should we check for exitOnEndofTopic as well?

I don't think we need to since even if "exitOnEndOfTopic" is set ,"isRunning" will still be set to true and any error will be bubbled up as expected

srkukarni · 2020-06-17T22:43:56Z

...ctions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java

-                            log.warn("Encountered error when assignment tailer is not running", th);
-                        }
-                    }
+        this.tailerThread = getTailerThread();


maybe defer this till start?

...unctions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java

srkukarni · 2020-06-17T22:50:41Z

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/LeaderService.java

+            try {
+                // trigger read to the end of the topic and exit
+                // Since the leader can just update its in memory assignments cache directly
+                functionRuntimeManager.stopReadingAssignments();


Can you instead call functionRunTimeManager.acquireLeadership() and functionRunTimeManager.giveupLeadership()

doesn't make to call "schedulerManager.initialize();" there or add the SchedulerManager as a dependency in FunctionRuntimeManager just for this

srkukarni · 2020-06-17T23:27:14Z

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/LeaderService.java

+            // when a worker has lost leadership it needs to start reading from the assignment topic again
+            try {
+                // acquire scheduler lock to make sure a scheduling is not in process
+                schedulerManager.getSchedulerLock().lock();


I think the better way is to make scheduler aware of the leadership changes(just like runtime manger) and call acquireLeadership and giveupLeadership

This is a way to do that. You will need synchronization somewhere and someone will have to wait

srkukarni · 2020-06-19T05:42:40Z

...ctions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java

    private volatile boolean isRunning = false;
+    private volatile boolean exitOnEndOfTopic = false;
+    private CompletableFuture<Void> hasExited;


exitFuture might be a better name

srkukarni · 2020-06-19T05:43:16Z

...ctions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java


-    private final Thread tailerThread;
+    @Getter
+    private MessageId lastMessageId = null;


Shouldn' t we init this to MessageId.earliest?

srkukarni · 2020-06-19T05:45:44Z

...ctions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java

-    public void start() {
-        isRunning = true;
-        tailerThread.start();
+    public synchronized void start() throws PulsarClientException {


I think its cleaner to consolidate this and above method to start(MessageId) { ... }

I also think that some logic will be simpler if we create Tailer object every time we go thru leadership transistion

I also think that some logic will be simpler if we create Tailer object every time we go thru leadership transition

That is not correct. The functionAssignmentTailer is also responsible for keeping track of a message id. If a worker becomes a leader and then loses leadership prior to creating any assignments, we shouldn't just start reading the assignment topic from the beginning. We should resume from the message id stored in the functionAssignmentTailer

srkukarni · 2020-06-19T05:48:58Z

...ctions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java

+                    if (msg == null) {
+                        if (exitOnEndOfTopic && !reader.hasMessageAvailable()) {
+                            break;
+                        }


is it simpler if we do
while(isRunning) {
if (exitOnEndOfTopic && !available) break;
try { read message... }

It's safer to wait for a timeout period to make sure no messages just arrived late

srkukarni · 2020-06-19T05:49:55Z

...nctions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionMetaDataManager.java


        } catch (Exception e) {
            log.error("Failed to initialize meta data store", e);
            throw new RuntimeException(e);
        }
    }
+
+    public void start() {


initialize and start? what cannot be done during constructor?

because we cannot start prior to the SchedulerManager is setup since function metadata manager can invoke the scheduler. We can initialize prior to to setting up the SchedulerManager but we cannot start

srkukarni · 2020-06-19T05:51:27Z

...unctions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java

-                    this.getWorkerService().getClient().newReader(),
-                    this.workerConfig,
-                    this.errorNotifier);
+


I like that we are no longer using FunctionAssignmenttailer here. However maybe a static method that consolidates. this reader creation and the one in assignment tailer?

srkukarni · 2020-06-19T05:57:19Z

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerUtils.java

@@ -33,8 +33,11 @@
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminBuilder;
 import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.MessageId;


* Fix leader/scheduler assignment processing lag problem * add license header * adding more comments * improving impl * fixing bugs * improving impl * fixing tests * adding comments * add more testing * addressing comments * cleaning up * refactoring implementation * addressing comments Co-authored-by: Jerry Peng <jerryp@splunk.com>

jerrypeng added the area/function label Jun 10, 2020

jerrypeng added this to the 2.6.0 milestone Jun 10, 2020

jerrypeng self-assigned this Jun 10, 2020

Fix leader/scheduler assignment processing lag problem

8a54e0f

jerrypeng force-pushed the function_scheduler_improvement branch from e8f550c to 8a54e0f Compare June 10, 2020 20:34

add license header

5406761

jerrypeng requested review from sijie, merlimat and srkukarni June 10, 2020 22:24

adding more comments

65c23f8

srkukarni reviewed Jun 12, 2020

View reviewed changes

Jerry Peng added 5 commits June 12, 2020 00:54

improving impl

66de46e

fixing bugs

4189de2

improving impl

9d2ec82

fixing tests

4557156

adding comments

85d3c0a

srkukarni reviewed Jun 14, 2020

View reviewed changes

add more testing

c10a521

codelipenghui modified the milestones: 2.6.0, 2.7.0 Jun 15, 2020

jerrypeng requested a review from srkukarni June 16, 2020 22:52

srkukarni reviewed Jun 17, 2020

View reviewed changes

...unctions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java Show resolved Hide resolved

srkukarni reviewed Jun 17, 2020

View reviewed changes

addressing comments

61f1c2b

Jerry Peng added 2 commits June 17, 2020 18:10

cleaning up

1c8f554

refactoring implementation

67c082e

srkukarni reviewed Jun 19, 2020

View reviewed changes

addressing comments

0d31bf5

srkukarni approved these changes Jun 19, 2020

View reviewed changes

srkukarni merged commit 68877f8 into apache:master Jun 19, 2020

Fix leader/scheduler assignment processing lag problem #7237

Fix leader/scheduler assignment processing lag problem #7237

Conversation

jerrypeng commented Jun 10, 2020

Motivation

Modifications

jerrypeng commented Jun 10, 2020

jerrypeng commented Jun 11, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerrypeng Jun 12, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerrypeng Jun 19, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerrypeng Jun 12, 2020 •

edited

Loading

jerrypeng Jun 19, 2020 •

edited

Loading