JFR profiling в Flink

Когда production Flink-job начинает работать медленно или нерегулярно, первый инструмент в руке инженера — Java Flight Recorder (JFR). Это встроенный в JVM profiler с уникальным свойством: его overhead настолько мал, что его можно запускать на работающей production-job без перерывов. Профилировщики типа YourKit или JProfiler требуют рестарта JVM и добавляют 10-30% overhead. JFR — 1-3% overhead и hot-pluggable.

Этот урок про практический workflow: как запустить JFR на работающем TaskManager, что собирать, как анализировать в JDK Mission Control (JMC), и какие выводы делать.

Что такое JFR

Java Flight Recorder — это event-based профилировщик, встроенный в Oracle JDK с Java 7 (и в OpenJDK с Java 11). Он записывает события JVM (method invocations, GC, allocations, lock contention, IO, network) в круговой buffer, который периодически дампится на диск.

Ключевое свойство — минимальный overhead. JFR не делает sampling каждые N инструкций, как другие profilers. Вместо этого он использует существующие JVM hooks: TLAB allocation, safepoint events, JIT compilation events. Большинство этих hooks JVM делает в любом случае — JFR просто записывает их.

Какие события записываются:

• Method profiling — sampling stack traces каждые ~10 ms.
• Allocation profiling — sampling object allocations (TLAB allocation events).
• GC events — все pauses, collection phases, heap statistics.
• Lock contention — threads waiting on monitors, RWLocks.
• IO events — file read/write, socket read/write.
• Class loading — каждый loaded класс.
• JIT compilation — methods being compiled, deopt events.

Overhead: типично 1-3% throughput drop на production workloads. Для коротких recordings (минуты) можно собирать полный профиль без видимого impact.

Запуск JFR на работающем Flink

Самый частый сценарий: production job работает, надо разобраться почему он медленный. Делаем dynamic JFR recording без рестарта.

Через jcmd (рекомендуется):

# Найти PID TaskManager
ps aux | grep TaskManagerRunner

# Запустить recording на 60 секунд
jcmd {PID} JFR.start name=flink_diag duration=60s \
    filename=/tmp/flink-{PID}.jfr settings=profile

# Status
jcmd {PID} JFR.check

# Принудительно остановить (если duration=0)
jcmd {PID} JFR.stop name=flink_diag

Settings:

• default — минимальный набор событий, низкий overhead (для baseline).
• profile — больше событий (recommended для диагностики).

Файл .jfr после завершения готов к анализу. Размер обычно 50-300 MB на минуту.

Через JVM startup flags (для всегда-включённого JFR):

# В flink-conf.yaml или TaskManager startup
env.java.opts: "-XX:+FlightRecorder -XX:StartFlightRecording=disk=true,maxsize=1g,maxage=1h,filename=/opt/flink/log/flink.jfr"

Это создаёт rolling recording: всегда last 1 hour доступен, max 1 GB. Когда инцидент случается, копируете снапшот и анализируете.

Анализ через JMC

JDK Mission Control (JMC) — это GUI для анализа JFR файлов. Бесплатный, скачивается отдельно.

# Установить JMC
brew install --cask jdk-mission-control  # macOS
# или скачать с openjdk.org/projects/jmc

# Открыть JFR
jmc -open flink-12345.jfr

JMC показывает несколько перспектив:

Automated Analysis: автоматический список проблем — GC overhead, lock contention, hot methods. Это первое что смотрите — JMC сам подсказывает, где смотреть глубже.

Method Profiling tab: flame graph и stack traces по hottest methods. Покажет, в каких классах/методах больше всего CPU.

Memory tab: allocation rate per class, heap usage timeline, GC pause analysis. Покажет какие классы аллоцируют больше всего.

Threads tab: per-thread CPU usage, blocking events, deadlock analysis. Покажет какие потоки заняты больше всего.

Hot methods analysis

Самый частый use case JFR — найти hot method. В JMC: Method Profiling -> Stack Traces View.

Что искать:

• Сериализация в hot path: если top frames это KryoSerializer.serialize, Output.writeBytes — у вас Kryo bottleneck (см. урок 17.2).
• State backend operations: если top — RocksDBValueState.update, EmbeddedRocksDBStateBackend.* — bottleneck в state.
• Network IO: если top — Bootstrap.run, EpollEventLoop — network bandwidth ограничение.
• User code: если top — ваш собственный код, профилируйте дальше его внутренности.

Пример типичного flame graph для CPU-bound Flink job:

Thread.run
└── TaskManagerRunner.run
    └── StreamTask.invoke
        └── OneInputStreamTask.processInput
            └── StreamOneInputProcessor.processInput
                ├── 65% - OperatorChain.process
                │   ├── 40% - MyMapFunction.map (USER CODE)
                │   └── 25% - RocksDBValueState.update
                ├── 20% - NetworkBuffer.allocate
                └── 15% - PojoSerializer.serialize

Здесь видим: user code занимает 40% — это разумно. RocksDB state 25% — высоковато, можно оптимизировать. Serialization 15% — приемлемо для POJO.

Allocation profiling

Allocation profiling в JFR показывает, какие классы создают больше всего объектов. Это критично для streaming jobs: каждый аллокированный объект eventually нужно GC-нить, что добавляет pauses.

В JMC: Memory -> Object Allocations.

Типичные паттерны:

1. Allocation hotspot — wrapper objects. Если top — java.lang.Long, java.lang.Integer — у вас auto-boxing в hot path. Часто происходит из-за Map{'<'}String, Long{'>'} без primitive map.

// ПЛОХО: auto-boxing на каждое event
Map{'<'}String, Long{'>'} counter = new HashMap{'<'}{'>'}();
counter.merge(userId, 1L, Long::sum);  // boxing Long вокруг 1L

// ХОРОШО: primitive map (Eclipse Collections или Koloboke)
MutableObjectLongMap{'<'}String{'>'} counter = ObjectLongHashMap.newMap();
counter.addToValue(userId, 1L);

2. Allocation hotspot — String. Если top — java.lang.String, byte[] (String internal) — у вас лишние String operations. Часто из StringBuilder без явной reuse:

// ПЛОХО: новый StringBuilder каждый раз
public String format(Event e) {
    return "user=" + e.user + ", amount=" + e.amount;
}

// ХОРОШО: переиспользовать через ThreadLocal
private static final ThreadLocal{'<'}StringBuilder{'>'} BUILDER =
    ThreadLocal.withInitial(() -{'>'} new StringBuilder(128));

public String format(Event e) {
    StringBuilder sb = BUILDER.get();
    sb.setLength(0);
    sb.append("user=").append(e.user).append(", amount=").append(e.amount);
    return sb.toString();
}

3. Allocation hotspot — Tuple/POJO. Это normal для Flink — каждое event это объект. Если allocation rate растёт линейно с throughput — это ожидаемо. Если расходится — есть лишние copies (см. enableObjectReuse, урок 17.4).

GC analysis

GC tab в JMC показывает полную картину сборок мусора: pause durations, allocation rate, generation sizes.

Что искать:

1. Frequent young GC (Minor GC). На healthy Flink job — Minor GC должен происходить 1-10 раз в секунду, каждый 5-50 ms. Если чаще — high allocation pressure, посмотрите allocation hotspots.

2. Long Full GC. Full GC на Flink job это плохой sign — означает что heap polluted long-lived объектами, которые не fit в young generation. Pauses 1+ second убивают latency. Решение: либо больше heap, либо уменьшить state-in-heap (переключиться на RocksDB state backend), либо найти leak.

3. GC overhead. JFR показывает % wall-clock time spent в GC. Healthy — under 5%. 10%+ — серьёзная проблема. 30%+ — job почти не работает, тратит CPU только на GC.

GC Statistics from JFR:
- Young Generation collections: 1247 (avg 18 ms, max 45 ms)
- Old Generation collections: 12 (avg 850 ms, max 2.1 s)
- Total time in GC: 23.4 seconds out of 5 minutes = 7.8% overhead
- Allocation rate: 3.2 GB/sec
- Promotion rate: 120 MB/sec

Это пример unhealthy stats:

• 7.8% GC overhead — выше комфорта (5%).
• Old GC занимает 0.8-2 секунды — критично для latency.
• Promotion rate 120 MB/sec означает много long-lived objects попадают в old generation — потенциальная leak или high state-in-heap.

Решения зависят от cause:

• Если allocation hotspots в user code — фиксить allocation (см. выше).
• Если state-in-heap big — переключиться на RocksDB.
• Если просто large state — увеличить heap (taskmanager.memory.task.heap.size).
• Если promotion rate high — настроить G1GC или ZGC (low-pause collectors).

Thread contention

Threads tab показывает per-thread CPU usage, blocking events, lock contention. Critical для finding parallelism issues.

Что искать:

1. Hot single thread. Если в TaskManager один thread занимает 95-100% CPU, а остальные idle — у вас skew или serial bottleneck.

2. Lock contention. Если в Thread CPU time большая доля в LockSupport.park, synchronized blocks — потоки ждут друг друга. Часто из-за глобального state в operator (singleton с synchronized).

3. IO wait. Если thread spends time в EpollEventLoop.processSelectedKeys — это network IO wait, normal. Если в RocksDB.write — RocksDB compaction backpressure (см. урок 19.3).

JFR пишет lock contention events с указанием owner thread, что упрощает разбор.

Custom JFR events

JFR можно использовать для custom events — например, профилировать конкретные операции в user code:

import jdk.jfr.Event;
import jdk.jfr.Label;
import jdk.jfr.Category;

@Label("Fraud Detection")
@Category("Application")
public class FraudDetectionEvent extends Event {
    @Label("User ID")
    public String userId;

    @Label("Decision")
    public String decision;

    @Label("Score")
    public double score;
}

// В user code
public class FraudDetector extends KeyedProcessFunction{'<'}...{'>'} {
    public void processElement(...) {
        FraudDetectionEvent event = new FraudDetectionEvent();
        event.begin();
        event.userId = userId;
        // ... actual work
        event.score = score;
        event.decision = decision;
        event.commit();
    }
}

Custom events появляются в JMC как отдельная category. Можно фильтровать, агрегировать, делать таймлайны. Полезно для domain-specific profiling — например measure latency per business operation.

JFR в Kubernetes

В K8s deployments PyFlink/Flink работают как pods. JFR remains accessible:

# Найти TaskManager pod
kubectl get pods -l app=flink-tm

# Запустить JFR через exec
kubectl exec -it flink-tm-0 -- jcmd 1 JFR.start \
    name=diag duration=60s filename=/opt/flink/log/diag.jfr

# Скопировать когда готово
kubectl cp flink-tm-0:/opt/flink/log/diag.jfr ./diag.jfr

Для регулярного collection — sidecar контейнер с cron-job, который дёргает jcmd и uploadит файлы в S3.

Production tip: настройте rolling JFR recording на старте каждого TM. Когда инцидент — у вас уже есть recording для предыдущего часа без необходимости что-то делать.

JFR vs async-profiler

Часто возникает вопрос: JFR или async-profiler? Они дополняют друг друга:

Recommendation:

• JFR для baseline + always-on recording. Это ваш default tool — оставьте rolling JFR включённым в production.
• async-profiler для conkretnyh investigations. Когда нужно глубже разобраться с CPU hotspot или allocation pattern — async-profiler даёт более точные flame graphs.

Async-profiler детально в следующем уроке.

Попробуй сам

1. JFR на dev TaskManager. Запустите Flink в local mode, jcmd JFR.start на 60 секунд. Откройте recording в JMC, изучите Automated Analysis вкладку.

2. Найти allocation hotspot. Симулируйте allocation issue: в user MapFunction создавайте новый StringBuilder каждый вызов. JFR должен показать его в top allocation hotspots.

3. Custom JFR event. Добавьте custom JFR event в ваш user code (например для measuring processing time per category). Откройте в JMC — увидите его как отдельный category.