sijie closed pull request #2223: fix bug in FunctionRuntimeManager involving
not cleaning up old invalid assignments
URL: https://github.com/apache/incubator-pulsar/pull/2223
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
index 05a79d8da5..5e1995e633 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
@@ -82,7 +82,6 @@
private MembershipManager membershipManager;
private final ConnectorsManager connectorsManager;
-
public FunctionRuntimeManager(WorkerConfig workerConfig,
PulsarClient pulsarClient,
Namespace dlogNamespace,
@@ -354,6 +353,7 @@ public synchronized void
removeAssignments(Collection<Assignment> assignments) {
* @param assignmentsUpdate the assignment update
*/
public synchronized void processAssignmentUpdate(MessageId messageId,
AssignmentsUpdate assignmentsUpdate) {
+
if (assignmentsUpdate.getVersion() > this.currentAssignmentVersion) {
Map<String, Assignment> assignmentMap = new HashMap<>();
@@ -430,8 +430,16 @@ public synchronized void processAssignmentUpdate(MessageId
messageId, Assignment
newFunctionRuntimeInfo.setFunctionInstance(assignment.getInstance());
this.insertStartAction(newFunctionRuntimeInfo);
this.setFunctionRuntimeInfo(fullyQualifiedInstanceId,
newFunctionRuntimeInfo);
- this.setAssignment(assignment);
}
+
+ // find existing assignment
+ Assignment existing_assignment =
this.findAssignment(assignment);
+ if (existing_assignment != null) {
+ // delete old assignment that could have old data
+ this.deleteAssignment(existing_assignment);
+ }
+ // set to newest assignment
+ this.setAssignment(assignment);
}
}
diff --git
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java
index 8ab7473eed..4f618c41b4 100644
---
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java
+++
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java
@@ -378,12 +378,10 @@ public boolean matches(Object o) {
.build()))));
Assert.assertEquals(functionRuntimeManager.functionRuntimeInfoMap.size(), 2);
-
-
Assert.assertEquals(functionRuntimeManager.workerIdToAssignments.size(), 2);
+
Assert.assertEquals(functionRuntimeManager.workerIdToAssignments.size(), 1);
Assert.assertEquals(functionRuntimeManager.workerIdToAssignments
.get("worker-1").get("test-tenant/test-namespace/func-1:0"),
assignment1);
Assert.assertEquals(functionRuntimeManager.workerIdToAssignments
.get("worker-1").get("test-tenant/test-namespace/func-2:0"),
assignment3);
}
-
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
