KAFKA-15022: [2/N] introduce graph to compute min cost by lihaosky · Pull Request #13996 · apache/kafka

lihaosky · 2023-07-11T22:28:55Z

Description

Introduce graph to calculate min-flow if an existing flow already input

Test

Unit test.

mjsax

Couple of initial comments.

Task :streams:checkstyleTest

[ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-13996/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/GraphTest.java:47:30: ',' is not followed by whitespace. [WhitespaceAfter]

[ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-13996/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/GraphTest.java:49:30: ',' is not followed by whitespace. [WhitespaceAfter]

[ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-13996/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/GraphTest.java:158:19: Variable 'exception' should be declared final. [FinalLocalVariable]

[ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-13996/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/GraphTest.java:169:19: Variable 'exception' should be declared final. [FinalLocalVariable]

[ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-13996/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/GraphTest.java:182:19: Variable 'exception' should be declared final. [FinalLocalVariable]

[ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-13996/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/GraphTest.java:194:19: Variable 'exception' should be declared final. [FinalLocalVariable]

[ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-13996/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/GraphTest.java:206:19: Variable 'exception' should be declared final. [FinalLocalVariable]

[ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-13996/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/GraphTest.java:218:19: Variable 'exception' should be declared final. [FinalLocalVariable]

mjsax · 2023-07-13T00:04:43Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

compareTo is to establish an order, right? Why do we order by (destination,capacity,cost); does is matter, or could we use any order as long as deterministic?

Any order should be fine. But as you pointed out. We can actually remove this.

lihaosky · 2023-07-13T05:22:23Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

I'll remove compareTo

mjsax · 2023-07-13T00:08:19Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

In Kafka, it's common practice to omit the get on getter, so it should just be nodes() (similar for other methods)

mjsax · 2023-07-13T00:10:18Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

Should this rather go into the constructor of Edge class?

mjsax · 2023-07-13T00:10:33Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

mjsax · 2023-07-13T00:18:06Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

Why do we use nested SortedMap? Is seems SortedMap<V, List<Edge> would be sufficient?

It's for getting the edge between two nodes efficiently. e.g. to get edge between node 0 and node 1, we can do adjList.get(0).get(1).

mjsax · 2023-07-13T00:19:51Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

Why do we implement Comparable? Could not spot the reason why it's required.

Good catch! I originally used SortedMap<V, SortedSet<Edge> as adjList which need to sort Edge. Later I changed it to SortedMap<V, SortedMap<V, Edge>> and we can remove this.

mjsax · 2023-07-13T19:30:56Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

Can you explain this condition?

This means their can't be a flow in this edge which essentially means these two nodes are disconnected.

mjsax

Few more questions.

mjsax · 2023-07-13T19:31:49Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

Looking into https://en.wikipedia.org/wiki/Bellman%E2%80%93Ford_algorithm that you linked from the KIP, the algorithms says "go over all edges" -- we need 2 for-loops to this but what is not totally obvious; might be worth to add a coment?

mjsax · 2023-07-13T19:32:10Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

From https://en.wikipedia.org/wiki/Bellman%E2%80%93Ford_algorithm it says, so it N-1 times, but we do it N times. Why?

Good question. N-1 times will find shortest path, we need to do it one more time to see if the path can be even shorter which mean there's a negative cycle. That's why there's a check on line 335.

mjsax · 2023-07-13T19:44:25Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

Can this be final? (cf comment below)

mjsax · 2023-07-13T19:44:52Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

Why do we need this update? Seems it would not change? Can we remove this line?

mjsax · 2023-07-13T19:45:54Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

Why do we need this update? Seems it would not change? Can we remove this line?

(We cannot make parentEdge final thought it seems, as it's set to a different value below)

Actually parentEdge is changing from line 283

mjsax · 2023-07-13T19:47:14Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

If I read this correctly, we are finding the minimum edge inside the cycle?

Yeah. If you want to flow through all the edges in the cycle, the maximum you can flow is the the minimum capacity (residualFlow) in all the edges.

mjsax · 2023-07-13T19:52:47Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

I was first confused about why we don't start with nodeInCycle and do everything is a single loop -- now I understand that we do it to construct an "exit" criteria -- seems worth to add a comment to explain it upfront.

mjsax · 2023-07-13T19:53:16Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

Not sure if I get this condition. Can you elaborate?

If there's some flow going in one direction, there should be the same amount of flow going in opposite direction.

In the case of forward edge, the counterEdge.flow is always larger than possibleFlow because forward edge's flow is backward edge's residual flow and possibleFlow is always smaller than or equal to residual flow.

In the case of backward edge, counterEdge.flow can be smaller than possibleFlow because forwardEdge's residual flow is bounded by capacity. Actually in case's counterEdge.flow < possibleFlow, we also need to reset it to 0. I'll fix that.

OK. After closer look, I think updating backward edge doesn't make much sense. So I added a forwardEdge bool in Edge to indicate whether it's forward edge and only update the flow in forward edge.

mjsax · 2023-07-13T19:55:18Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

Not sure what this part does.

This is to update the original graph edge to the computed flow. Basically iterating all edges in original graph and change the flow to what's in residualGraph's corresponding edge.

mjsax

Couple is smaller comments/questions. Can be addressed by follow up PR (or if you want inside N/3).

mjsax · 2023-07-17T23:19:50Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

+        int totalCost = 0;
+        for (final Map.Entry<V, SortedMap<V, Edge>> nodeEdges : adjList.entrySet()) {
+            final SortedMap<V, Edge> edges = nodeEdges.getValue();
+            for (final Entry<V, Edge> nodeEdge : edges.entrySet()) {


nit: we could just iterate over the valueSet ?

mjsax · 2023-07-17T23:20:55Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

+        }
+
+        public Edge(final V destination, final int capacity, final int cost, final int residualFlow, final int flow,
+            final boolean forwardEdge) {


nit: formatting (if it does not fit in one line, we should move each parameter into it's one line to simplify reading)

public Edge( final V destination, final int capacity, final int cost, final int residualFlow, final int flow, final boolean forwardEdge ) {

mjsax · 2023-07-17T23:21:57Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

+
+            final Graph<?>.Edge otherEdge = (Graph<?>.Edge) other;
+
+            return destination.equals(otherEdge.destination) && capacity == otherEdge.capacity


nit formatting:

return destination.equals(otherEdge.destination) && capacity == otherEdge.capacity && cost == otherEdge.cost && residualFlow == otherEdge.residualFlow && flow == otherEdge.flow && forwardEdge == otherEdge.forwardEdge;

mjsax · 2023-07-17T23:22:44Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

+
+        @Override
+        public String toString() {
+            return "{destination= " + destination + ", capacity=" + capacity + ", cost=" + cost


return "Edge: {...}" ?

mjsax · 2023-07-17T23:23:26Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

+        @Override
+        public String toString() {
+            return "{destination= " + destination + ", capacity=" + capacity + ", cost=" + cost
+                + ", residualFlow=" + residualFlow + ", flow=" + flow + ", forwardEdge=" + forwardEdge;


missing }.

Should we also switch to one line per parameter we print to simplify reading?

mjsax · 2023-07-20T22:49:55Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/GraphTest.java

+        graph.addEdge(4, 2, 1, 0, 1);
+        graph.addEdge(1, 5, 1, 0, 1);
+        graph.addEdge(3, 5, 1, 0, 1);
+        graph.setSourceNode(4);


Would it be simpler to use 0 as source? (At least for my mind it's easier to follow what going on, if numbers are "ordered")

Or to avoid a lot of re-writing, name the source -1 (and the sink 99) so both a clearly different, and we don't need to update too much code.

Sounds good. Will use -1 and 99 then

mjsax · 2023-07-20T23:05:08Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java

+        final V destination;
+        final int capacity;
+        final int cost;
+        int residualFlow;


Do we actually need this field?

If I read the code right, for a forward edge, it's always capacity - flow and for a backward edge it's always 0. So it seem redundant (and potentially error prone to store it expliclity)? -- Instead we could have a residualFlow() method that compute it on-the-fly (we could also simplify the update logic when modifying flow as we only need to update the flow itself)? Or do I read the update logic inside cancelNegativeCycle incorrectly and those properties are not an invariant?

for a backward edge it's always 0

residualFlow for backward edge is the flow in forward edge: https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java#L178

Ah. Parameter order... Thought it's , flow, residualFlow, (not sure why you picked the "reverse" order)

mjsax · 2023-07-20T23:13:14Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/GraphTest.java

+        edge = edges.get(1);
+        assertEquals(1, edge.flow);
+        assertEquals(0, edge.residualFlow);
+    }


Seems we do not check all conditions? Flow from 0->1 is shifted to 0->3, right?

0->3 flow is checked on line 269?

mjsax · 2023-07-20T23:13:45Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/GraphTest.java

+        graph1.addEdge(4, 6, 1, 0, 1);
+
+        graph1.setSourceNode(5);
+        graph1.setSinkNode(6);


mjsax · 2023-07-20T23:15:21Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/GraphTest.java

+        edgeList.add(new TestEdge(4, 5, 1, 1, 0));
+
+        // Test no matter the order of adding edges, min cost flow flows from 0 to 2 and then from 2 to 5
+        for (int i = 0; i < 10; i++) {


Why do we need to test this 10 times? Given that the test runs for each PR on nightly builds, it seems sufficient to just run it once?

Want to test it more if someone trigger this manually. I feel 10 times won't hurt much, but I'm fine switching to 1 if you feel strongly 😄

Test should run fast. Don't feel strongly about it.

Part of KIP-925. Reviewers: Matthias J. Sax <matthias@confluent.io>

mjsax added streams kip Requires or implements a KIP labels Jul 12, 2023

mjsax reviewed Jul 13, 2023

View reviewed changes

lihaosky added 8 commits July 17, 2023 11:59

KAFKA-15022: [2/N] introduce graph to compute min cost

bea537e

More tests and validation

12c18a9

comments

f73051e

More test

f26c021

checkstyle

1b5556d

comments

4ca93ec

comments

9e67923

change addEdge to private

738b345

lihaosky mentioned this pull request Jul 17, 2023

KAFKA-15022: [3/N] use graph to compute rack aware assignment for active stateful tasks #14030

Merged

lihaosky force-pushed the min-cost-graph branch from 344f54e to 738b345 Compare July 17, 2023 22:22

bug fix

a2889a4

mjsax reviewed Jul 20, 2023

View reviewed changes

mjsax merged commit 6bb88ae into apache:trunk Jul 20, 2023

jeqo pushed a commit to jeqo/kafka that referenced this pull request Jul 21, 2023

KAFKA-15022: [2/N] introduce graph to compute min cost (apache#13996)

c3379ed

Part of KIP-925. Reviewers: Matthias J. Sax <matthias@confluent.io>

Cerchie pushed a commit to Cerchie/kafka that referenced this pull request Jul 25, 2023

KAFKA-15022: [2/N] introduce graph to compute min cost (apache#13996)

5445062

Part of KIP-925. Reviewers: Matthias J. Sax <matthias@confluent.io>

lihaosky deleted the min-cost-graph branch August 8, 2023 17:09

jeqo pushed a commit to aiven/kafka that referenced this pull request Aug 15, 2023

KAFKA-15022: [2/N] introduce graph to compute min cost (apache#13996)

57f91a3

Part of KIP-925. Reviewers: Matthias J. Sax <matthias@confluent.io>


		final Graph<?>.Edge otherEdge = (Graph<?>.Edge) other;

		return destination.equals(otherEdge.destination) && capacity == otherEdge.capacity

Conversation

lihaosky commented Jul 11, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Test

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This comment was marked as resolved.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

lihaosky commented Jul 11, 2023 •

edited

Loading

mjsax Jul 20, 2023 •

edited

Loading