What are the API calls made to create snapshot while using RocksDB as the state backend in Flink?

I want to profile the time taken for each of the API calls that Flink makes to the RocksDB. However, I am unable to find those functions.

I have tried setting up the complete source code of Flink in an IDE, integrated my streaming example in the source code, started the debugger and stepped into many of the calls made, but in vain.

Here is the example:

package org.apache.flink.streaming.examples.spendreport;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;
import org.apache.flink.walkthrough.common.sink.AlertSink;
import org.apache.flink.walkthrough.common.source.TransactionSource;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;

/**
 * Normal code.
 */
public class FraudDetectionAvi {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		DataStream<Transaction> transactions = env
			.addSource(new TransactionSource())
			.name("transactions");

		env.enableCheckpointing(60000);
		env.setStateBackend(new RocksDBStateBackend("file:///home/avsrivas/dev/flink/checkpoints", true));

		DataStream<Alert> alerts = transactions
			.keyBy(Transaction::getAccountId)
			.process(new FraudDetectorAvi())
			.name("fraud-detector");

		alerts
			.addSink(new AlertSink())
			.name("send-alerts");

		env.execute("Fraud Detection");
	}
}

Click here for the complete source code.

I tried stepping into the execute but I could not deduce the functions that make the call to RocksDB for saving the state.

When RocksDB is used as the state backend for a Flink application, then the working copy of any key-partitioned state is stored in a local, embedded RocksDB instance in each task manager. Timers may kept there as well, or they may be on the heap. RocksDB keeps its state on the local disk; non-keyed state is always on the heap.

When a snapshot is taken (i.e., during checkpointing or when taking a savepoint), the state stored in RocksDB is (asynchronously) copied to the snapshot storage (which should be a distributed filesystem).

In your application, when you call flagState.update(true), for example, that ends up here, in RocksDBValueState.java, which uses this code to write to RocksDB:

backend.db.put(columnFamily, writeOptions,
	serializeCurrentKeyWithGroupAndNamespace(),
	serializeValue(value));

What happens later, during snapshotting, depends on whether you use incremental or full checkpointing, but you'll find the RocksDB-specific code in https://github.com/kebab-mai-haddi/flink/tree/master/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/snapshot.

Note that the snapshots are not stored in RocksDB. Incremental snapshots are taken by mirroring the SST files, and full snapshots involve iterating over all of the state in the state backend and writing out the result.

For more on how Flink uses RocksDB, search for blog posts and Flink Forward talks by Stefan Richter.