[Flink-39911][historyserver] Support Pluggable Storage Backend for HistoryServer by chenzihao5 · Pull Request #28402 · apache/flink

chenzihao5 · 2026-06-12T06:05:28Z

What is the purpose of the change

This pull request introduces a pluggable archive storage backend for the HistoryServer, so that downloaded job archives no longer have to be unpacked into a large number of small JSON files on the local filesystem. And this PR adds a RocksDB backend that stores all archive entries as key-value pairs inside a single embedded RocksDB instance, while keeping the existing FILE backend as the default to preserve backwards compatibility.

Brief change log

Introduce the ArchiveStorage abstraction with two implementations:
- FileArchiveStorage — existing behavior, one JSON file per entry under historyserver.web.tmpdir (default).
- RocksDBArchiveStorage — stores all entries as key-value pairs in a single embedded RocksDB instance under historyserver.web.tmpdir/rocksdb.
Add new ConfigOptions in HistoryServerOptions:
- historyserver.archive.storage.type (FILE | ROCKSDB, default FILE)
- historyserver.archive.rocksdb.native-lib-dir
- historyserver.archive.rocksdb.compression
- historyserver.archive.rocksdb.bottommost-compression
Wire the selected backend into HistoryServer / HistoryServerArchiveFetcher so archive read/write paths go through ArchiveStorage.
Standardized the resource request processing workflow in AbstractHistoryServerHandler.
Add documents for this feature.

Verifying this change

This change added tests and can be verified as follows:

Added unit tests for ArchiveStorage covering exists / get / put / delete / deletePrefix / getByPrefix and lifecycle of archive data.
Added common unit tests in AbstractHistoryServerHandlerTest for different archive storage.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): yes
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? docs

Was generative AI tooling used to co-author this PR?

Yes (Claude-4.7-Opus 1M Context)

flinkbot · 2026-06-12T06:09:48Z

CI report:

b99c7ec Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

reswqa

Thanks @chenzihao5 for improving history server.

I made a rough scan first and left some comments.

… FileArchiveStorage implementation

…m HistoryServerStaticFileServerHandler

chenzihao5 · 2026-06-16T07:57:17Z

@reswqa Thanks for your review. I have modified the code based on the review comments. Please take a look again. Thanks.

reswqa · 2026-06-17T07:37:21Z

+    void deleteByPrefix(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<Entry> getEntriesByPrefix(String prefix) throws IOException;


I suggest either keeping all the Entries word from method name or removing them all.

reswqa · 2026-06-17T07:38:38Z

+     *
+     * @param prefix storage key prefix
+     * @return the entries
+     * @throws IOException if the entries cannot be read


@throws IOException if the entries cannot be read

Any entries or all?

reswqa · 2026-06-17T07:48:25Z

+
+    @Override
+    public boolean exists(String key) throws IOException {
+        return new File(rootPath, key).exists();


I wonder could we avoid the allocation here(e.g. using Files.exists)?

reswqa · 2026-06-17T07:49:02Z

+        File target = new File(rootPath, key);
+        Files.deleteIfExists(target.toPath());


Could we avoid creating the File instance here?

reswqa · 2026-06-17T07:53:54Z

        private final FileSystem fs;

-        private RefreshLocation(Path path, FileSystem fs) {
+        protected RefreshLocation(Path path, FileSystem fs) {


Why this should be protected?

reswqa · 2026-06-17T07:58:14Z

 * HistoryServerOptions#HISTORY_SERVER_CLEANUP_EXPIRED_APPLICATIONS}.
 */
-public class HistoryServerApplicationArchiveFetcher extends HistoryServerArchiveFetcher {
+public class HistoryServerApplicationArchiveFetcher<Entry>


Add java doc for the generic type Entry.

reswqa · 2026-06-17T07:59:58Z

+        File webApplicationDir = new File(webDir, APPLICATIONS_SUBDIR);
        Files.createDirectories(webApplicationDir.toPath());
-        this.webApplicationsOverviewDir = new File(webDir, APPLICATION_OVERVIEWS_SUBDIR);
+        File webApplicationsOverviewDir = new File(webDir, APPLICATION_OVERVIEWS_SUBDIR);
        Files.createDirectories(webApplicationsOverviewDir.toPath());


It seems that we need the Path rather than File obj itselft.

reswqa · 2026-06-17T08:10:17Z

+            String key;
            if (path.equals(JobsOverviewHeaders.URL)) {
-                target = new File(webOverviewDir, jobId + JSON_FILE_ENDING);
+                key = "/" + JOB_OVERVIEWS_SUBDIR + "/" + jobId + JSON_FILE_ENDING;


I suggest we extract "/" + JOB_OVERVIEWS_SUBDIR + "/" and "/" + JOBS_SUBDIR + "/" as a const String variable(e.g. JOB_OVERVIEWS_KEY_PREFIX and JOBS_KEY_PREFIX or any other reasonable name).

reswqa · 2026-06-17T08:13:10Z

+                if (overview instanceof File) {
+                    subJobs = mapper.readValue((File) overview, MultipleJobsDetails.class);
+                } else {
+                    subJobs = mapper.readValue((String) overview, MultipleJobsDetails.class);
                }


We assume that Entry either a File or a String, If I introduce a new type of ArchivedStore then things got worse. I don't think dev should add new branch here?

reswqa · 2026-06-17T09:03:27Z

+    protected abstract ArchiveStorage<T> createStorage() throws Exception;
+
+    /** Reads the textual content of a storage entry. */
+    protected abstract String readContent(T entry) throws Exception;


If we introduce a method like getAsContent in the ArchivedStore that directly returns the string, then this method and the generic type of the test class would no longer be necessary. Furthermore, we should be able to convert it into parameterized test class.

reswqa · 2026-06-17T09:08:28Z

+        final Path uploadDir = Files.createDirectory(tmpDir.resolve("uploadDir"));
+
+        AbstractHistoryServerHandler<?> handler = createHandler(webDir.toFile());
+        Router router = new Router().addGet("/:*", handler);


Raw use of parameterized class 'Router'

reswqa · 2026-06-17T09:08:57Z

+ * Common HTTP-level tests for {@link AbstractHistoryServerHandler} subclasses. Subclasses only need
+ * to provide a concrete handler instance via {@link #createHandler(File)}.
+ */
+abstract class AbstractHistoryServerHandlerTest {


This could be a parameterized test.

reswqa · 2026-06-17T09:14:06Z

+                                    + "extracts the library into a unique sub-directory under this "
+                                    + "directory. Defaults to the JVM 'java.io.tmpdir' when not "
+                                    + "configured. Only applies when "
+                                    + "'historyserver.archive.storage.type' is 'ROCKSDB'.");


I prefer HISTORY_SERVER_ARCHIVE_STORAGE_TYPE.key() rather than hard-coded historyserver.archive.storage.type.

reswqa · 2026-06-17T09:19:04Z

+    public static final ConfigOption<RocksDBCompressionType>
+            HISTORY_SERVER_ARCHIVE_ROCKSDB_COMPRESSION =
+                    key("historyserver.archive.rocksdb.compression")
+                            .enumType(RocksDBCompressionType.class)
+                            .defaultValue(RocksDBCompressionType.LZ4_COMPRESSION)
+                            .withDescription(
+                                    "Compression type used for the non-bottommost levels of the RocksDB "
+                                            + "SST files. Only applies when 'historyserver.archive.storage.type' is 'ROCKSDB'.");
+
+    /** Compression type used for the bottommost level of the RocksDB SST files. */
+    public static final ConfigOption<RocksDBCompressionType>
+            HISTORY_SERVER_ARCHIVE_ROCKSDB_BOTTOMMOST_COMPRESSION =
+                    key("historyserver.archive.rocksdb.bottommost-compression")
+                            .enumType(RocksDBCompressionType.class)
+                            .defaultValue(RocksDBCompressionType.ZSTD_COMPRESSION)
+                            .withDescription(
+                                    "Compression type used for the bottommost level of the "
+                                            + "RocksDB SST files. Only applies when 'historyserver.archive.storage.type' is 'ROCKSDB'.");


These configurations are too low-level. Do any users actually have the need to optimize these? If so, it's not too late to introduce them later. I prefer to remove these two configuration options first.

reswqa · 2026-06-17T11:15:01Z

+            byte[] startKey = keyPrefix.getBytes(UTF_8);
+            byte[] endKey = keyPrefix.getBytes(UTF_8);
+            // Add 1 to the last byte to get the next lexicographic byte
+            endKey[endKey.length - 1]++;


IIUC, this trick should heavily rely on the fact that the byte array is encoded in UTF-8. It would be best to add some comments to explain why it works.

reswqa · 2026-06-17T11:18:03Z

+    @Override
+    public List<String> getEntriesByPrefix(String prefix) throws IOException {
+        List<String> result = new ArrayList<>();
+        if (prefix == null || prefix.isEmpty()) {


StringUtils.isNullOrWhitespaceOnly(prefix))

reswqa · 2026-06-17T11:24:53Z

+                } else {
+                    break;
+                }
+            }


Perhaps we should add a check after while block per https://github.com/facebook/rocksdb/wiki/Iterator 🤔

while(iterator.isValid()){ xxx } iterator.status();

reswqa · 2026-06-17T11:30:11Z

+                    libDir.getAbsolutePath());
+        } catch (Throwable t) {
+            LOG.warn("Failed to load RocksDB native library to '{}'.", libDir.getAbsolutePath(), t);
+            deleteDirectoryQuietly(libDir);


Should we clean up this libDir even if history server stop w/o error?

reswqa · 2026-06-17T11:51:41Z

+        try (RocksIterator iterator = db.newIterator()) {
+            byte[] prefixBytes = prefix.getBytes(UTF_8);
+            iterator.seek(prefixBytes);
+            while (iterator.isValid()) {


Could we use ReadOptions.setIterateUpperBound(Slice) to set upbound here?

byte[] upper = xxx; try (Slice upperSlice = new Slice(upper); ReadOptions ro = new ReadOptions().setIterateUpperBound(upperSlice); RocksIterator it = db.newIterator(ro)) { for (it.seek(prefixBytes); it.isValid(); it.next()) { result.add(new String(it.value(), UTF_8)); } try { it.status(); } catch (RocksDBException e) { throw new IOException(e); } }

reswqa · 2026-06-18T02:49:57Z

+
+    @Override
+    public void close() {
+        handlesToClose.forEach(IOUtils::closeQuietly);


I think we should close db first. The db internally holds references to objects from options — including the bloom filter. As long as the db is alive, these resources are still in use by background threads, flushes, and compactions. If you free the options before closing the db, it may accesses freed memory(i.e. use-after-free).

Even without this issue, generally speaking, the rule-of-thumb is close all the handlesToClose in reverse order.

reswqa reviewed Jun 12, 2026

View reviewed changes

chenzihao5 added 2 commits June 16, 2026 13:01

[FLINK-39911][historyserver] Introduce ArchiveStorage abstraction and…

9e24cf2

… FileArchiveStorage implementation

[FLINK-39911][historyserver] Extract AbstractHistoryServerHandler fro…

4d89643

…m HistoryServerStaticFileServerHandler

chenzihao5 force-pushed the FLINK-39911 branch from 2f340b5 to 0c2ce50 Compare June 16, 2026 07:48

chenzihao5 added 2 commits June 16, 2026 16:20

[FLINK-39911][historyserver] Support RocksDB as archive storage backend

89df37d

[FLINK-39911][docs] Document HistoryServer archive storage backend

b99c7ec

chenzihao5 force-pushed the FLINK-39911 branch from 0c2ce50 to b99c7ec Compare June 16, 2026 08:25

reswqa reviewed Jun 17, 2026

View reviewed changes

reswqa reviewed Jun 18, 2026

View reviewed changes

		File target = new File(rootPath, key);
		Files.deleteIfExists(target.toPath());

Conversation

chenzihao5 commented Jun 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

flinkbot commented Jun 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

reswqa left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

chenzihao5 commented Jun 16, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

reswqa Jun 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

reswqa Jun 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

chenzihao5 commented Jun 12, 2026 •

edited

Loading

flinkbot commented Jun 12, 2026 •

edited

Loading

reswqa Jun 17, 2026 •

edited

Loading

reswqa Jun 18, 2026 •

edited

Loading