State Backend Configuration

/**
 * RocksDB state backend that stores state in embedded RocksDB instances.
 * This is the recommended state backend for production workloads requiring durable state.
 */
class EmbeddedRocksDBStateBackend extends AbstractManagedMemoryStateBackend implements ConfigurableStateBackend {
    
    /**
     * Creates a new EmbeddedRocksDBStateBackend with undefined incremental checkpointing.
     * The incremental checkpointing setting will be determined by configuration.
     */
    EmbeddedRocksDBStateBackend();
    
    /**
     * Creates a new EmbeddedRocksDBStateBackend with specified incremental checkpointing.
     * @param enableIncrementalCheckpointing whether to enable incremental checkpointing
     */
    EmbeddedRocksDBStateBackend(boolean enableIncrementalCheckpointing);
    
    /**
     * Creates a new EmbeddedRocksDBStateBackend with ternary incremental checkpointing setting.
     * @param enableIncrementalCheckpointing incremental checkpointing setting (true/false/undefined)
     */
    EmbeddedRocksDBStateBackend(TernaryBoolean enableIncrementalCheckpointing);
}

// Basic setup with incremental checkpointing enabled
EmbeddedRocksDBStateBackend stateBackend = new EmbeddedRocksDBStateBackend(true);

// Setup with configuration-determined incremental checkpointing
EmbeddedRocksDBStateBackend stateBackend = new EmbeddedRocksDBStateBackend();

/**
 * Sets a single storage path for RocksDB files.
 * @param path directory path for RocksDB storage
 */
void setDbStoragePath(String path);

/**
 * Sets multiple storage paths for RocksDB files to distribute I/O load.
 * @param paths array of directory paths for RocksDB storage
 */
void setDbStoragePaths(String... paths);

/**
 * Gets the configured storage paths.
 * @return array of configured storage paths
 */
String[] getDbStoragePaths();

// Single storage path
stateBackend.setDbStoragePath("/data/flink/rocksdb");

// Multiple paths for I/O distribution
stateBackend.setDbStoragePaths(
    "/data1/flink/rocksdb", 
    "/data2/flink/rocksdb", 
    "/data3/flink/rocksdb"
);

/**
 * Sets predefined RocksDB options optimized for specific hardware profiles.
 * @param options predefined configuration set
 */
void setPredefinedOptions(PredefinedOptions options);

/**
 * Gets the current predefined options configuration.
 * @return current predefined options
 */
PredefinedOptions getPredefinedOptions();

/**
 * Sets a custom RocksDB options factory for fine-grained configuration.
 * @param optionsFactory factory for creating RocksDB options
 */
void setRocksDBOptions(RocksDBOptionsFactory optionsFactory);

/**
 * Gets the current RocksDB options factory.
 * @return current options factory
 */
RocksDBOptionsFactory getRocksDBOptions();

// Use predefined options
stateBackend.setPredefinedOptions(PredefinedOptions.FLASH_SSD_OPTIMIZED);

// Use custom options factory
DefaultConfigurableOptionsFactory factory = new DefaultConfigurableOptionsFactory()
    .setMaxBackgroundThreads(4)
    .setWriteBufferSize("128mb");
stateBackend.setRocksDBOptions(factory);

/**
 * Sets the number of threads for file transfer during checkpointing.
 * @param numberOfTransferThreads thread count for parallel file transfer
 */
void setNumberOfTransferThreads(int numberOfTransferThreads);

/**
 * Gets the configured number of transfer threads.
 * @return number of transfer threads
 */
int getNumberOfTransferThreads();

/**
 * Sets the maximum size of write batches for RocksDB operations.
 * @param writeBatchSize maximum write batch size in bytes
 */
void setWriteBatchSize(long writeBatchSize);

/**
 * Gets the configured write batch size.
 * @return write batch size in bytes, or -1 if not set
 */
long getWriteBatchSize();

// Configure checkpoint transfer threads
stateBackend.setNumberOfTransferThreads(8);

// Configure write batch size (64MB)
stateBackend.setWriteBatchSize(64 * 1024 * 1024);

/**
 * Sets the priority queue state type for timer service.
 * @param priorityQueueStateType implementation type for priority queues
 */
void setPriorityQueueStateType(PriorityQueueStateType priorityQueueStateType);

/**
 * Gets the configured priority queue state type.
 * @return current priority queue state type
 */
PriorityQueueStateType getPriorityQueueStateType();

// Use heap-based priority queues (faster, limited by memory)
stateBackend.setPriorityQueueStateType(PriorityQueueStateType.HEAP);

// Use RocksDB-based priority queues (scalable beyond memory limits)  
stateBackend.setPriorityQueueStateType(PriorityQueueStateType.ROCKSDB);

/**
 * Checks if incremental checkpointing is enabled.
 * @return true if incremental checkpointing is enabled
 */
boolean isIncrementalCheckpointsEnabled();

/**
 * Creates a configured copy of the state backend from ReadableConfig.
 * @param config configuration to apply
 * @param classLoader class loader for loading factory classes
 * @return new configured state backend instance
 */
EmbeddedRocksDBStateBackend configure(ReadableConfig config, ClassLoader classLoader);

/**
 * Gets the memory configuration for this state backend.
 * @return memory configuration object
 */
RocksDBMemoryConfiguration getMemoryConfiguration();

// Check incremental checkpointing status
if (stateBackend.isIncrementalCheckpointsEnabled()) {
    System.out.println("Incremental checkpointing is enabled");
}

// Access memory configuration
RocksDBMemoryConfiguration memConfig = stateBackend.getMemoryConfiguration();
memConfig.setWriteBufferRatio(0.4);

enum PriorityQueueStateType {
    /** Heap-based priority queue (faster access, limited by memory) */
    HEAP,
    
    /** RocksDB-based priority queue (scalable beyond memory limits) */
    ROCKSDB
}

enum TernaryBoolean {
    TRUE,
    FALSE,
    UNDEFINED
}