Narrow down the scope of waiting for pending tasks to per partition #191

ta7uw · 2023-03-02T01:35:04Z

Motivation:

Described in #172

Modifications:

Reload context per partition when reloading is requested

Result:

Closes Narrow the scope of waiting pending task count upon dynamic property reload #172
Reduces time to resume processing when dynamic property reloaded

ocadaruma

Thanks for submitting the PR.

This PR seems trying to observe pending task count per-partition.
However, it's not enough because as long as reloadRequested is true, Decaton pauses all processing so anyways none of partitions can make progress. (

decaton/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContexts.java

Line 215 in c4bbbc1

    
           return processingRateProp.value() == RateLimiter.PAUSED || reloadRequested.get();

)

So we need to manage reload-requested flag per-partition I think

ta7uw · 2023-03-06T00:43:32Z

Thanks for your review.
I'll take it into consideration.

ta7uw · 2023-03-06T08:56:09Z

@ocadaruma
Could you please review it again?

ocadaruma · 2023-03-07T01:51:39Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContext.java

+    public void reloading(boolean reloading) {
+        this.reloading = reloading;
+        if (reloading) {
+            pause();


pausing here by PartitionContext itself doesn't work to pause fetching actually.

Decaton's fetch-pause works like:

(1) ConsumeManager retrieves partition list that needs pause https://github.com/line/decaton/blob/v6.2.0/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ConsumeManager.java#L182

(2) Call Consumer#pause to stop fetching further records. From now on, Consumer#poll doesn't return records for these partitions

So, if we pause here by PartitionContext itself, it will not be included in (1) (because it's already "paused").

Correct approach here may be holding per-partition reloading flag in PartitionContexts, and include them in partitionsNeedsPause

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContexts.java

ta7uw · 2023-03-07T04:42:14Z

@ocadaruma
I fixed to hold reload request per-partition.
Could you please review it again?

ocadaruma · 2023-03-11T23:42:07Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContexts.java

@@ -53,6 +57,8 @@ public class PartitionContexts implements OffsetsStore, AssignmentStore, Partiti
    private final Map<TopicPartition, PartitionContext> contexts;

    private final AtomicBoolean reloadRequested;


Do we still need this reloadRequested flag? Since we manage reloading states in reloadStates hash map

reloadRequested is useful for quickly determining if contexts should be reloaded without iteration.

Hm, IMO having duplicated info (with per-partition reload) could be an error prone when we maintain this code in the future, while iterating over partitions unlikely becomes a bottleneck (since we already doing iterating partitions for updating high watermarks, checking partition needs pause/resume) so the benefit to have reloadRequested separately is small.

ocadaruma · 2023-03-12T00:22:44Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContexts.java

+                List<TopicPartition> reloadableTopicPartitions = reloadStates.entrySet()
+                        .stream()
+                        .filter(entry -> entry.getValue()
+                                         && contexts.get(entry.getKey()).pendingTasksCount() == 0)


This could cause NPE because currently reloadStates entry never removed even when partition is removed.

To fix that, we can add a code to remove reloadStates entry at the same time we modify contexts.

However, then, we have to maintain two separate Map which holds per-partition states, which is bothersome and an error-prone.

So I think we should reloadRequested flag in PartitionContext itself. WDYT

I got the point.
Having a reloadRequested flag in PartitionContext seems good.

ocadaruma · 2023-03-12T00:26:06Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContexts.java

@@ -214,7 +233,9 @@ public List<TopicPartition> partitionsNeedsPause() {
        return contexts.values().stream()
                       .filter(c -> !c.revoking())
                       .filter(c -> !c.paused())
-                       .filter(c -> pausingAll || shouldPartitionPaused(c.pendingTasksCount()))
+                       .filter(c -> pausingAll
+                                    || reloadStates.get(c.topicPartition())


Additionally, we also need to modify partitionNeedsResume to avoid returning partitions that reload is requested, because otherwise, partitions will be resumed immediately so new tasks will be kept feeding so pendingTasks never becomes 0.

ocadaruma

Left few comments but almost looks good.

Also would you mind adding an integration test to check changing the concurrency in the middle of processing doesn't cause problem? (referring RateLimiterTest might be helpful)

ocadaruma · 2023-03-23T14:11:24Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContext.java

+     */
+    @Getter
+    @Setter
+    private boolean reloadState;


reloadRequested sounds more clear?

Should be marked as volatile to ensure latest value (set by property-listener thread) is visible to subscription thread

ocadaruma · 2023-03-23T14:29:01Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContexts.java

@@ -52,7 +56,7 @@ public class PartitionContexts implements OffsetsStore, AssignmentStore, Partiti
    private final int maxPendingRecords;
    private final Map<TopicPartition, PartitionContext> contexts;

-    private final AtomicBoolean reloadRequested;
+    private final ReentrantLock lock;


Let's rename to something like propertyReloadLock for understandability

ocadaruma · 2023-03-23T14:36:32Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContexts.java

-            if (!reloadRequested.getAndSet(true)) {
+            lock.lock();
+            try {
+                contexts.values().forEach(context -> context.reloadState(true));


To prevent contexts are modified during the iteration on this property-listener thread, we need to surround all places that maintaining contexts entry with lock. (currently only maybeHandlePropertyReload is surrounded, but there are several places modifying contexts)

Do you mean having another ReentrantLock for updating the contexts?

No, use propertyReloadLock.

ocadaruma

Left last few nits comments!

Also could you fix this comment?

decaton/processor/src/main/java/com/linecorp/decaton/processor/runtime/ProcessorProperties.java

Line 77 in 5659ec7

    
                * Reloading this property will pause all assigned partitions until current pending tasks have done.

as reload is now done in per-partition

ocadaruma · 2023-03-27T00:38:22Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContexts.java

+                    .filter(entry -> entry.getValue().reloadRequested()
+                                     && entry.getValue().pendingTasksCount() == 0)
+                    .map(Entry::getKey)
+                    .collect(Collectors.toList());


[nits] toList is already static-imported so let's omit Collectors

ocadaruma · 2023-03-27T00:38:47Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContexts.java

+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
+import java.util.stream.Collectors;


[nits] There are several unused imports

ocadaruma · 2023-03-27T00:41:07Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContexts.java

-
-        logger.info("Start dropping partition contexts");
+    private void reloadContexts(Collection<TopicPartition> topicPartitions) {
+        logger.info("Start dropping partition context({})", topicPartitions);
        removePartition(topicPartitions);
        logger.info("Finished dropping partition contexts. Start recreating partition contexts");


[nits] This log and this line

decaton/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContexts.java

Line 316 in 5659ec7

logger.info("Completed reloading property");

should also include topicPartitions

ocadaruma

LGTM. Nice!

Narrow down the scope of waiting for pending tasks to per partition

c4bbbc1

ocadaruma reviewed Mar 2, 2023

View reviewed changes

Manage reloading request per-partition

fe7f2be

ocadaruma reviewed Mar 7, 2023

View reviewed changes

Hold reload state per-partition

90bc937

Fix

24ebdef

ocadaruma added new feature Add a new feature and removed new feature Add a new feature labels Mar 10, 2023

ocadaruma self-requested a review March 10, 2023 02:57

ocadaruma reviewed Mar 12, 2023

View reviewed changes

Address the comments by @ocadaruma

b55869f

ocadaruma self-requested a review March 23, 2023 09:29

ocadaruma reviewed Mar 23, 2023

View reviewed changes

ta7uw added 2 commits March 24, 2023 16:11

Rename property and Lock contexts

aa59181

Add integration test

5659ec7

ocadaruma reviewed Mar 27, 2023

View reviewed changes

Address the comment by @ocadaruma

929bcc5

ocadaruma approved these changes Mar 27, 2023

View reviewed changes

ocadaruma merged commit c95626a into line:master Mar 27, 2023

ta7uw deleted the issue-172 branch March 27, 2023 02:31

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Narrow down the scope of waiting for pending tasks to per partition #191

Narrow down the scope of waiting for pending tasks to per partition #191

ta7uw commented Mar 2, 2023

ocadaruma left a comment

ta7uw commented Mar 6, 2023

ta7uw commented Mar 6, 2023

ocadaruma Mar 7, 2023

ta7uw commented Mar 7, 2023

ocadaruma Mar 11, 2023

ta7uw Mar 17, 2023

ocadaruma Mar 21, 2023

ocadaruma Mar 12, 2023

ta7uw Mar 17, 2023

ocadaruma Mar 12, 2023

ocadaruma left a comment

ocadaruma Mar 23, 2023

ocadaruma Mar 23, 2023

ocadaruma Mar 23, 2023

ta7uw Mar 24, 2023

ocadaruma Mar 24, 2023

ocadaruma left a comment

ocadaruma Mar 27, 2023

ocadaruma Mar 27, 2023

ocadaruma Mar 27, 2023

ocadaruma left a comment

		@@ -53,6 +57,8 @@ public class PartitionContexts implements OffsetsStore, AssignmentStore, Partiti
		private final Map<TopicPartition, PartitionContext> contexts;

		private final AtomicBoolean reloadRequested;

Narrow down the scope of waiting for pending tasks to per partition #191

Narrow down the scope of waiting for pending tasks to per partition #191

Conversation

ta7uw commented Mar 2, 2023

ocadaruma left a comment

Choose a reason for hiding this comment

ta7uw commented Mar 6, 2023

ta7uw commented Mar 6, 2023

Choose a reason for hiding this comment

ta7uw commented Mar 7, 2023

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

ocadaruma left a 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

ocadaruma left a 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

ocadaruma left a comment

Choose a reason for hiding this comment