DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
Motivation
In large-scale production environments, the HistoryServer often needs to manage a very high number of completed Flink jobs. Each job archive typically contains many small files, including job metadata, task-level details, and so on. When both the number of jobs and the volume of archived details are excessive, a large number of small files are generated, consuming an excessive number of inodes.
So I want to propose introducing RocksDB as a local storage backend to reduce small files, while keeping file-based storage unchanged.
- Benefit: Mitigate issues with small files and prevent the system inode exhaustion.
In my local testing environment, which involves over 20 jobs, the default storage type(FILE) generates more than 400,000 archive files, whereas the RocksDB storage type requires only about 20 files.
FILE | ROCKSDB |
> 40w | ≈ 20 |
Public Interfaces
ArchiveStorage
Introduce ArchiveStorage interface for pluggable storage backends.
/**
* Abstraction for the storage backend used by the {@link HistoryServer} to read archived job data.
*
* <p>Implementations can be backed by the local file system, RocksDB, or any other storage medium.
*
* @param <T> Type of the storage entries.
*/
public interface ArchiveStorage<T> {
/**
* Returns whether the entry identified by {@code key} exists in this storage.
*
* @param key storage key (typically the request path, e.g. {@code /jobs/xxx/config.json})
* @return {@code true} if the entry exists
*/
boolean exists(String key);
/**
* Returns the entry identified by {@code key} from this storage.
*
* @param key storage key
* @return the entry, or null if the entry does not exist
* @throws IOException if the entry cannot be read
*/
@Nullable
T get(String key) throws IOException;
/**
* Stores the entry identified by {@code key} in this storage.
*
* @param key storage key
* @param archiveContent the archive content to store, this type is string because the archive content is always a JSON String
* @throws IOException if the entry cannot be written
*/
void put(String key, String archiveContent) throws IOException;
/**
* Deletes the entry identified by {@code key} from this storage.
*
* @param key storage key
* @throws IOException if the entry cannot be deleted
*/
void delete(String key) throws IOException;
/**
* Deletes all entries with key starting with {@code keyPrefix} from this storage.
*
* <p>Such as deleting all archived files for a given job or application.
*
* @param keyPrefix key prefix
* @throws IOException if entries cannot be deleted
*/
void deletePrefix(String keyPrefix) throws IOException;
/**
* Returns the entries identified by {@code prefix} from this storage.
*
* @param prefix storage key prefix
* @return the entries
* @throws IOException if the entries cannot be read
*/
List<T> getByPrefix(String prefix) throws IOException;
Proposed Changes
Summary
ArchiveStorage
├── FileArchiveStorage (Storage Type: FILE)
└── RocksDBArchiveStorage (Storage Type: ROCKSDB)
AbstractHistoryServerHandler (Abstract base class; standardizes processes such as resource loading and response handling)
├── HistoryServerStaticFileServerHandler (Storage Type: FILE)
└── HistoryServerRocksDBHandler (Storage Type: ROCKSDB)
1. Configuration
Key | Default | Type | Description |
historyserver.archive.storage.type | FILE | Enum | Define how to store the archive locally:
|
2. Resource Handling
Introduce AbstractHistoryServerHandler standardizes the request processing workflow into two steps: loadResource and respondWithResource.
- Static Files(such index.html, xx.js …): Accessed directly by the ClassLoader.
- Archive Files: Load from the specified archive storage.
- HistoryServerStaticFileServerHandler for FILE archive storage
- HistoryServerRocksDBHandler for ROCKSDB archive storage
public abstract class AbstractHistoryServerHandler<T>
extends SimpleChannelInboundHandler<RoutedRequest> {
...
protected void respondToRequest(ChannelHandlerContext ctx, RoutedRequest routedRequest)
throws Exception {
...
// Archive Files
if (!requestPath.contains(".")) {
Resource resource = loadResource(path);
respondWithResource(ctx, routedRequest.getRequest(), requestPath, resource);
} else {
// Static Files
tryLoadFromClassloader(destFile, requestPath);
responseWithFile(ctx, routedRequest.getRequest(), requestPath, destFile);
}
}
/** Respond to the request with the resource. */
protected abstract void respondWithResource(
ChannelHandlerContext ctx, HttpRequest request, String requestPath, T resource)
throws Exception;
/**
* Loads the resource for the given request path from the archive storage.
*
* @param requestPath The request path
* @return The resource for the given request path, or null if not found
*/
private T loadResource(String requestPath) throws Exception {
if (archiveStorage.exists(requestPath)) {
return archiveStorage.get(requestPath);
}
return null;
}
...
}
3. Archive Fetcher
Change relevant File operations to archiveStorage operations. such as:
// before File target = new File(webDir, path + JSON_FILE_ENDING); writeTargetFile(target, json); // after String key = path + JSON_FILE_ENDING; archiveStorage.put(key, json);
Compatibility, Deprecation, and Migration Plan
The default storage remains FILE, ensuring backward compatibility.
The ROCKSDB storage will be validated for 1–2 releases. If stable, they may become the default.
Test Plan
- Add unit tests for new functionality
- Perform end-to-end testing with a local HistoryServer deployment
Rejected Alternatives
N/A