Flink: throttle messages to walkaround deadlock under heavy iteration feedback load

Flink: throttle messages to walkaround deadlock under heavy iteration feedback load.md

Flink has a deadlock case and can be caused by the combination of Flink's backpressure mechanism with iterations (more likely when there is heavy feedback load).

There is a proposal to resolve this issue, but it seems to be abondoned.

The walkaround, suggested by Gábor Hermann, is to throttle messages into the iteration.

Here is my implementation.

elementsPerSecond is a hyper parameter that needs to be tuned manually and it seems to be related with operator network structure. Put your network under high pressure, especially heavy feedback load. If deadlocks happens, half elementsPerSecond and try again.

Hope it helps.

ThrottledMapper.java

package com.ybsjyyn.bdc.logistics.operators;

import org.apache.flink.api.common.functions.MapFunction;

/**
 * Throttle messages
 * <p>
 * ref: https://github.com/apache/flink/blob/master/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/utils/ThrottledIterator.java
 */
public class ThrottledMapper<T> implements MapFunction<T, T> {
    private final long sleepBatchSize;
    private final long sleepBatchTime;

    private long lastBatchCheckTime;
    private long num;

    public ThrottledMapper(long elementsPerSecond) {
        if (elementsPerSecond >= 100) {
            // how many elements would we emit per 50ms
            this.sleepBatchSize = elementsPerSecond / 20;
            this.sleepBatchTime = 50;
        } else if (elementsPerSecond >= 1) {
            // how long does element take
            this.sleepBatchSize = 1;
            this.sleepBatchTime = 1000 / elementsPerSecond;
        } else {
            throw new IllegalArgumentException(
                    "'elements per second' must be positive and not zero");
        }

        this.lastBatchCheckTime = System.currentTimeMillis();
    }

    @Override
    public T map(T event) throws Exception {
        num++;
        if (num < sleepBatchSize) {
            lastBatchCheckTime = System.currentTimeMillis();
            return event;
        }

        num = 0;
        final long elapsed = System.currentTimeMillis() - lastBatchCheckTime;
        if (elapsed < sleepBatchTime) {
            try {
                Thread.sleep(sleepBatchTime - elapsed);
            } catch (InterruptedException e) {
                // restore interrupt flag and proceed
                Thread.currentThread().interrupt();
            }
        }

        lastBatchCheckTime = System.currentTimeMillis();
        return event;
    }
}