Приостановка обработки на Flink KeyedStream

Этого можно добиться с помощью Оператор окна Flink. Во-первых, создайте поток на основе POJO или tuple, применив операцию map.

Затем, в соответствии с вашими потребностями, вы можете использовать keyBy в этом потоке, чтобы получить keyedStream.

Теперь, используя комбинацию основанных на времени бесконечных window, trigger и window function, вы можете добиться переключения вашего командного потока.

По сути, вы можете использовать windows в качестве своего буфера, который после получения записи о паузе хранит записи процесса до тех пор, пока не будет получена запись возобновления. Вы бы написали настраиваемый триггер, который вытесняет окно (буфер) в соответствии с вашим сценарием.

Ниже приведена индивидуальная реализация Trigger с переопределенным методом onElement().

/**
 * We trigger the window processing as per command inside the record. The
 * process records are buffered when a pause record is received and the
 * buffer is evicted once resume record is received. If no pause record is
 * received earlier, then for each process record the buffer is evicted.
 */
@Override
public TriggerResult onElement(Tuple2<Integer, String> element, long timestamp, Window window,
        TriggerContext context) throws Exception {
    if (element.f1.equals("pause")) {
        paused = true;
        return TriggerResult.CONTINUE;
    } else if (element.f1.equals("resume")) {
        paused = false;
        return TriggerResult.FIRE_AND_PURGE;
    } else if (paused) // paused is a ValueState per keyed stream.
        return TriggerResult.CONTINUE;
    return TriggerResult.FIRE_AND_PURGE;
}

Посмотрите полный рабочий пример в этом репозиторий github