private void populateInOutFlow(final Map<V, Long> inFlow, final Map<V, Long> outFlow) {
        for (final Entry<V, SortedMap<V, Edge>> nodeEdges : adjList.entrySet()) {
            final V node = nodeEdges.getKey();
            if (node.equals(sinkNode)) {
                throw new IllegalStateException("Sink node " + sinkNode + " shouldn't have output");
            }
            for (final Entry<V, Edge> nodeEdge : nodeEdges.getValue().entrySet()) {
                final V destination = nodeEdge.getKey();
                if (destination.equals(sourceNode)) {
                    throw new IllegalStateException("Source node " + sourceNode + " shouldn't have input " + node);
                }
                final Edge edge = nodeEdge.getValue();
                Long count = outFlow.get(node);
                if (count == null) {
                    outFlow.put(node, (long) edge.flow);
                } else {
                    outFlow.put(node, count + edge.flow);
                }

                count = inFlow.get(destination);
                if (count == null) {
                    inFlow.put(destination, (long) edge.flow);
                } else {
                    inFlow.put(destination, count + edge.flow);
                }
            }
        }
    }
Solve min cost flow with cycle canceling algorithm.
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/Graph.java