[FLINK-40256][docs] Include sub-package config options in the configuration reference

docs/layouts/shortcodes/generated/expert_rocksdb_section.html

@@ -20,6 +20,48 @@
             <td>String</td>
             <td>The local directory (on the TaskManager) where RocksDB puts its files. Per default, it will be &lt;WORKING_DIR&gt;/tmp. See <code class="highlighter-rouge">process.taskmanager.working-dir</code> for more details.</td>
         </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-auto-compactions</h5></td>
+            <td style="word-wrap: break-word;">30</td>
+            <td>Integer</td>
+            <td>The maximum number of automatic compactions running for manual compaction to start. If the actual number is higher, manual compaction won't be started to avoid delaying automatic ones.</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-file-size-to-compact</h5></td>
+            <td style="word-wrap: break-word;">50 kb</td>
+            <td>MemorySize</td>
+            <td>The maximum size of individual input files</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-files-to-compact</h5></td>
+            <td style="word-wrap: break-word;">30</td>
+            <td>Integer</td>
+            <td>The maximum number of input files to compact together in a single compaction run</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-output-file-size</h5></td>
+            <td style="word-wrap: break-word;">64 mb</td>
+            <td>MemorySize</td>
+            <td>The maximum output file size</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-parallel-compactions</h5></td>
+            <td style="word-wrap: break-word;">5</td>
+            <td>Integer</td>
+            <td>The maximum number of manual compactions to start. Note that only one of them can run at a time as of v8.10.0; all the others will be waiting</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.min-files-to-compact</h5></td>
+            <td style="word-wrap: break-word;">5</td>
+            <td>Integer</td>
+            <td>The minimum number of input files to compact together in a single compaction run</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.min-interval</h5></td>
+            <td style="word-wrap: break-word;">0 ms</td>
+            <td>Duration</td>
+            <td>The minimum interval between manual compactions. Zero disables manual compactions</td>
+        </tr>
         <tr>
             <td><h5>state.backend.rocksdb.options-factory</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

docs/layouts/shortcodes/generated/rocksdb_manual_compaction_configuration.html

@@ -0,0 +1,54 @@
+<table class="configuration table table-bordered">
+    <thead>
+        <tr>
+            <th class="text-left" style="width: 20%">Key</th>
+            <th class="text-left" style="width: 15%">Default</th>
+            <th class="text-left" style="width: 10%">Type</th>
+            <th class="text-left" style="width: 55%">Description</th>
+        </tr>
+    </thead>
+    <tbody>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-auto-compactions</h5></td>
+            <td style="word-wrap: break-word;">30</td>
+            <td>Integer</td>
+            <td>The maximum number of automatic compactions running for manual compaction to start. If the actual number is higher, manual compaction won't be started to avoid delaying automatic ones.</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-file-size-to-compact</h5></td>
+            <td style="word-wrap: break-word;">50 kb</td>
+            <td>MemorySize</td>
+            <td>The maximum size of individual input files</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-files-to-compact</h5></td>
+            <td style="word-wrap: break-word;">30</td>
+            <td>Integer</td>
+            <td>The maximum number of input files to compact together in a single compaction run</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-output-file-size</h5></td>
+            <td style="word-wrap: break-word;">64 mb</td>
+            <td>MemorySize</td>
+            <td>The maximum output file size</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.max-parallel-compactions</h5></td>
+            <td style="word-wrap: break-word;">5</td>
+            <td>Integer</td>
+            <td>The maximum number of manual compactions to start. Note that only one of them can run at a time as of v8.10.0; all the others will be waiting</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.min-files-to-compact</h5></td>
+            <td style="word-wrap: break-word;">5</td>
+            <td>Integer</td>
+            <td>The minimum number of input files to compact together in a single compaction run</td>
+        </tr>
+        <tr>
+            <td><h5>state.backend.rocksdb.manual-compaction.min-interval</h5></td>
+            <td style="word-wrap: break-word;">0 ms</td>
+            <td>Duration</td>
+            <td>The minimum interval between manual compactions. Zero disables manual compactions</td>
+        </tr>
+    </tbody>
+</table>

flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java

@@ -68,6 +68,9 @@ public class ConfigurationOptionLocator {
                 new OptionsClassLocation(
                         "flink-state-backends/flink-statebackend-rocksdb",
                         "org.apache.flink.state.rocksdb"),
+                new OptionsClassLocation(
+                        "flink-state-backends/flink-statebackend-rocksdb",
+                        "org.apache.flink.state.rocksdb.sstmerge"),
                 new OptionsClassLocation(
                         "flink-state-backends/flink-statebackend-forst",
                         "org.apache.flink.state.forst"),
@@ -132,6 +135,11 @@ public ConfigurationOptionLocator(OptionsClassLocation[] locations, String pathP
         this.pathPrefix = pathPrefix;
     }
 
+    @VisibleForTesting
+    static OptionsClassLocation[] getLocations() {
+        return LOCATIONS;
+    }
+
     public void discoverOptionsAndApply(
             Path rootDir,
             BiConsumerWithException<Class<?>, Collection<OptionWithMetaInfo>, ? extends Exception>

flink-docs/src/test/java/org/apache/flink/docs/util/ConfigurationOptionLocatorTest.java

@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.docs.util;
+
+import org.apache.flink.annotation.docs.Documentation;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ConfigurationOptionLocator}. */
+class ConfigurationOptionLocatorTest {
+
+    private static final String SOURCE_ROOT = "src/main/java";
+
+    private static final String SECTION_ANNOTATION = "@Documentation.Section";
+
+    private static final Set<String> PRUNED_DIRECTORIES =
+            Collections.unmodifiableSet(
+                    new HashSet<>(Arrays.asList("target", "node_modules", ".git")));
+
+    /** Mirrors the file names {@link ConfigurationOptionLocator} recognizes. */
+    private static final Pattern OPTIONS_CLASS_FILE_NAME =
+            Pattern.compile("[a-zA-Z]*(?:Options|Config|Parameters)\\.java");
+
+    /**
+     * Verifies that every option annotated with {@link Documentation.Section} sits in a package
+     * that {@link ConfigurationOptionLocator} actually searches.
+     *
+     * <p>The annotation is an explicit statement that the option belongs in the generated
+     * configuration reference, but discovery is driven by a hard-coded list of packages and does
+     * not recurse into sub-packages. An option outside that list is therefore dropped from the
+     * reference without any error, and {@code ConfigOptionsDocsCompletenessITCase} cannot catch it
+     * because it derives its expectations from the same list.
+     */
+    @Test
+    void testSectionAnnotatedOptionsAreAllDiscoverable() throws IOException {
+        final Path rootDir = Paths.get(Utils.getProjectRootDir()).toAbsolutePath().normalize();
+
+        final Set<String> searchedPackages =
+                Arrays.stream(ConfigurationOptionLocator.getLocations())
+                        .map(
+                                location ->
+                                        location.getModule()
+                                                + '/'
+                                                + location.getPackage().replace('.', '/'))
+                        .collect(Collectors.toSet());
+
+        final List<String> undiscoverable = new ArrayList<>();
+        for (Path optionsClass : findSectionAnnotatedOptionClasses(rootDir)) {
+            final String relativePath = toUnixPath(rootDir.relativize(optionsClass));
+            final String modulePath = relativePath.substring(0, relativePath.indexOf(SOURCE_ROOT));
+            final String packagePath =
+                    relativePath.substring(
+                            modulePath.length() + SOURCE_ROOT.length() + 1,
+                            relativePath.lastIndexOf('/'));
+
+            if (!searchedPackages.contains(modulePath + packagePath)) {
+                undiscoverable.add(relativePath);
+            }
+        }
+
+        assertThat(undiscoverable)
+                .as(
+                        "The options in these classes are annotated with @Documentation.Section but "
+                                + "cannot be found by %s, so they are silently missing from the "
+                                + "generated configuration reference. Add an %s entry for the "
+                                + "containing package to %s#LOCATIONS.",
+                        ConfigurationOptionLocator.class.getSimpleName(),
+                        OptionsClassLocation.class.getSimpleName(),
+                        ConfigurationOptionLocator.class.getSimpleName())
+                .isEmpty();
+    }
+
+    private static List<Path> findSectionAnnotatedOptionClasses(Path rootDir) throws IOException {
+        final List<Path> optionClasses = new ArrayList<>();
+
+        Files.walkFileTree(
+                rootDir,
+                new SimpleFileVisitor<Path>() {
+                    @Override
+                    public FileVisitResult preVisitDirectory(
+                            Path dir, BasicFileAttributes attributes) {
+                        return PRUNED_DIRECTORIES.contains(dir.getFileName().toString())
+                                ? FileVisitResult.SKIP_SUBTREE
+                                : FileVisitResult.CONTINUE;
+                    }
+
+                    @Override
+                    public FileVisitResult visitFile(Path file, BasicFileAttributes attributes)
+                            throws IOException {
+                        if (OPTIONS_CLASS_FILE_NAME.matcher(file.getFileName().toString()).matches()
+                                && toUnixPath(file).contains('/' + SOURCE_ROOT + '/')
+                                && isSectionAnnotated(file)) {
+                            optionClasses.add(file);
+                        }
+                        return FileVisitResult.CONTINUE;
+                    }
+                });
+
+        return optionClasses;
+    }
+
+    private static boolean isSectionAnnotated(Path file) throws IOException {
+        try (Stream<String> lines = Files.lines(file)) {
+            return lines.anyMatch(line -> line.contains(SECTION_ANNOTATION));
+        }
+    }
+
+    private static String toUnixPath(Path path) {
+        return path.toString().replace(path.getFileSystem().getSeparator(), "/");
+    }
+}

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/sstmerge/RocksDBManualCompactionOptions.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.state.rocksdb.sstmerge;
 
+import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.annotation.docs.Documentation;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
@@ -26,6 +27,7 @@
 import java.time.Duration;
 
 /** Configuration options for manual compaction for the RocksDB backend. */
+@PublicEvolving
 public class RocksDBManualCompactionOptions {
 
     @Documentation.Section(Documentation.Sections.EXPERT_ROCKSDB)
@@ -42,7 +44,7 @@ public class RocksDBManualCompactionOptions {
                     .intType()
                     .defaultValue(5)
                     .withDescription(
-                            "The maximum number of manual compactions to start."
+                            "The maximum number of manual compactions to start. "
                                     + "Note that only one of them can run at a time as of v8.10.0; all the others will be waiting");
 
     @Documentation.Section(Documentation.Sections.EXPERT_ROCKSDB)
@@ -81,6 +83,6 @@ public class RocksDBManualCompactionOptions {
                     .intType()
                     .defaultValue(30)
                     .withDescription(
-                            "The maximum number of automatic compactions running for manual compaction to start."
+                            "The maximum number of automatic compactions running for manual compaction to start. "
                                     + "If the actual number is higher, manual compaction won't be started to avoid delaying automatic ones.");
 }