Replacing Coordinator Queue With Deque & Fixing Usage Of toMap Util #950

shrinandthakkar · 2023-07-24T17:19:33Z

Summary

Deque Coordinator Queue

When a newly elected leader receives an instance-task assignment from ZK, it is expected that the leader will perform a cleanup routine on the assignment. Failure to do so may lead to inconsistencies in task assignment across instances.
In the event that the cleanup routine encounters an exception, the "LEADER_DO_ASSIGNMENT" event with the "isNewlyElectedLeader" flag enabled must be reinserted at the front of the coordinator queue to ensure that no other events are processed before re-handling the event.

Bug Fixing toMap() Usage

Replace the use of Collectors.toMap(keyMapper, valueMapper) in cases where duplicate values may be encountered, with Collectors.toMap(keyMapper, valueMapper, mergeFunction).

Important: DO NOT REPORT SECURITY ISSUES DIRECTLY ON GITHUB.
For reporting security issues and contributing security fixes,
please, email security@linkedin.com instead, as described in
the contribution guidelines.

Please, take a minute to review the contribution guidelines at:
https://github.com/linkedin/Brooklin/blob/master/CONTRIBUTING.md

jzakaryan · 2023-07-25T22:03:25Z

datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java

    _leaderDoAssignmentScheduledFuture = _scheduledExecutor.schedule(() -> {
-      _eventQueue.put(CoordinatorEvent.createLeaderDoAssignmentEvent(isNewlyElectedLeader));
+      _eventQueue.put(CoordinatorEvent.createLeaderDoAssignmentEvent(isNewlyElectedLeader), false);


This is the main change in this PR. When a newly elected leader does assignment, it schedules the assignment event with task cleanup in front of the queue, to make sure it gets executed before anything else. Just to confirm my understanding.

Correct! We just need to make sure that nothing else gets handled apart from a successful handling of "LEADER_DO_ASSIGNMENT" (with newly elected leader flag enabled) for a newly elected leader.

isNewlyElectedLeader this will always be true, right? or when do we have that false?

isNewlyElectedLeader is only true when a new leader is elected. In many other cases we have to ask the leader to do assignments, and at those calls, isNewlyElectedLeader will be false.

jzakaryan · 2023-07-25T22:26:09Z

...m-server-restli/src/test/java/com/linkedin/datastream/server/CallableCoordinatorForTest.java

+/**
+ * Callable Coordinator is used for overriding coordinator behaviors for tests
+ */
+public interface CallableCoordinatorForTest {


Why do we need this interface? Seems like TestCoordinator.java has all you need.

I didnt wanted to add another constructor in the TestCoordinator.java to override another method of coordinator. With this interface, we could minimize code duplication and pass the overrides of coordinator as an argument.

For the test "testLeaderDoAssignmentForNewlyElectedLeaderFailurePath", I overrode performPreAssignmentCleanup method to test a failure path, where I am using this.

ehoner · 2023-08-02T15:39:57Z

...tream-server/src/main/java/com/linkedin/datastream/server/CoordinatorEventBlockingQueue.java

+
+  /**
+   * Add a single event to the queue, de-duping events with the same name and same metadata.
+   * @param event CoordinatorEvent event to add to the queue
+   * @param insertInTheEnd if true, indicates to add the event to the end of the queue and front, otherwise.
+   */
+  public synchronized void put(CoordinatorEvent event, boolean insertInTheEnd) {


Can the public APIs be similar to a Deque? I think that put(<event>) and putFirst(<event>) are clearer than using a boolean flag. The boolean option is fine for the private internal implementation.

...c/main/java/com/linkedin/datastream/server/assignment/StickyPartitionAssignmentStrategy.java

datastream-server-restli/src/test/java/com/linkedin/datastream/server/TestCoordinator.java

hshukla · 2023-08-02T17:36:04Z

...tream-server/src/main/java/com/linkedin/datastream/server/CoordinatorEventBlockingQueue.java

   * @param event CoordinatorEvent event to add to the queue
   */
  public synchronized void put(CoordinatorEvent event) {
-    LOG.info("Queuing event {} to event queue", event.getType());
+    put(event, true);


When do we need to support for inserting at rear?

All the events to this coordinator queue are inserted in the rear. The only case in which we have to insert in the front is what the PR proposes.

hshukla · 2023-08-02T17:47:38Z

...tream-server/src/main/java/com/linkedin/datastream/server/CoordinatorEventBlockingQueue.java

+   * @param event CoordinatorEvent event to add to the queue
+   * @param insertInTheEnd if true, indicates to add the event to the end of the queue and front, otherwise.
+   */
+  public synchronized void put(CoordinatorEvent event, boolean insertInTheEnd) {


for understanding: LinkedBlockingDeque is already a thread safe and we are using only offer or offerFirst, what is the rational on having this method synchronized?

[Minor]: another thing is, we want to have add from both ends but remove should be only from front. However, by using deque we will allow add/remove from both ends, Is there a way we can restrict remove from rear?

The synchronized methods were added long before but I am guessing that the shared set variable might create race and would let in duplicate events in some cases if those methods are not synchronized.

That is why the wrapper on top of the LinkedBlockingDeque is implemented which only supports taking out events from the front.

…oordinator Queue

jzakaryan

lgtm

...tream-server/src/main/java/com/linkedin/datastream/server/CoordinatorEventBlockingQueue.java

datastream-server-restli/src/test/java/com/linkedin/datastream/server/TestCoordinator.java

shrinandthakkar force-pushed the fixPartitionValidationErrorBMM branch from 8c79c2c to c1dfa84 Compare July 24, 2023 18:07

Replacing Coordinator Queue With Deque & Fixing Usage Of toMap Util

625fb0b

shrinandthakkar force-pushed the fixPartitionValidationErrorBMM branch from c1dfa84 to 625fb0b Compare July 24, 2023 22:05

shrinandthakkar marked this pull request as ready for review July 24, 2023 22:06

shrinandthakkar requested review from jzakaryan and ehoner July 24, 2023 22:06

jzakaryan requested changes Jul 25, 2023

View reviewed changes

shrinandthakkar requested a review from jzakaryan July 26, 2023 00:19

jzakaryan previously approved these changes Aug 1, 2023

View reviewed changes

ehoner requested changes Aug 2, 2023

View reviewed changes

hshukla reviewed Aug 2, 2023

View reviewed changes

shrinandthakkar dismissed jzakaryan’s stale review via e122778 August 3, 2023 16:32

Modify the public facing api for putting events in the front of the C…

a7f4e03

…oordinator Queue

shrinandthakkar force-pushed the fixPartitionValidationErrorBMM branch from e122778 to a7f4e03 Compare August 4, 2023 04:34

jzakaryan self-requested a review August 7, 2023 23:51

jzakaryan previously approved these changes Aug 7, 2023

View reviewed changes

ehoner reviewed Aug 8, 2023

View reviewed changes

...tream-server/src/main/java/com/linkedin/datastream/server/CoordinatorEventBlockingQueue.java Show resolved Hide resolved

datastream-server-restli/src/test/java/com/linkedin/datastream/server/TestCoordinator.java Outdated Show resolved Hide resolved

Better Handling Duplicate putFirst Requests & Added couple more tests

e7f3e2d

shrinandthakkar dismissed jzakaryan’s stale review via e7f3e2d August 8, 2023 19:13

ehoner requested changes Aug 8, 2023

View reviewed changes

datastream-server-restli/src/test/java/com/linkedin/datastream/server/TestCoordinator.java Outdated Show resolved Hide resolved

ehoner previously approved these changes Aug 8, 2023

View reviewed changes

hshukla previously approved these changes Aug 8, 2023

View reviewed changes

Fixing Flaky Test

b8d7192

shrinandthakkar dismissed stale reviews from hshukla and ehoner via b8d7192 August 8, 2023 20:45

ehoner approved these changes Aug 8, 2023

View reviewed changes

hshukla approved these changes Aug 8, 2023

View reviewed changes

shrinandthakkar merged commit d19a2c2 into linkedin:master Aug 8, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Replacing Coordinator Queue With Deque & Fixing Usage Of toMap Util #950

Replacing Coordinator Queue With Deque & Fixing Usage Of toMap Util #950

shrinandthakkar commented Jul 24, 2023 •

edited

Loading

jzakaryan Jul 25, 2023

shrinandthakkar Jul 26, 2023

hshukla Aug 2, 2023

shrinandthakkar Aug 2, 2023

jzakaryan Jul 25, 2023

shrinandthakkar Jul 26, 2023

ehoner Aug 2, 2023

hshukla Aug 2, 2023

shrinandthakkar Aug 2, 2023

shrinandthakkar Aug 3, 2023

hshukla Aug 2, 2023

shrinandthakkar Aug 2, 2023

hshukla Aug 2, 2023

hshukla Aug 2, 2023

shrinandthakkar Aug 2, 2023

jzakaryan left a comment

Replacing Coordinator Queue With Deque & Fixing Usage Of toMap Util #950

Replacing Coordinator Queue With Deque & Fixing Usage Of toMap Util #950

Conversation

shrinandthakkar commented Jul 24, 2023 • edited Loading

Summary

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

jzakaryan left a comment

Choose a reason for hiding this comment

shrinandthakkar commented Jul 24, 2023 •

edited

Loading