Repository: risingwavelabs/risingwave
File: java/connector-node/risingwave-source-cdc/src/main/java/com/risingwave/connector/cdc/debezium/internal/OpendalSchemaHistory.java (Java)

Code before:
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class OpendalSchemaHistory extends AbstractFileBasedSchemaHistory {
    private static final Logger LOGGER = LoggerFactory.getLogger(OpendalSchemaHistory.class);
    private String sourceId = "";
    public static final String SOURCE_ID = "schema.history.internal.source.id";
    public static final String MAX_RECORDS_PER_FILE_CONFIG =
            "schema.history.internal.max.records.per.file";
    private int maxRecordsPerFile = 2048; // default records nums per file
    private String objectDir = "";

    private static final Pattern HISTORY_FILE_PATTERN =
            Pattern.compile("schema_history_(\\d+)\\.dat");

    // Cache the latest file information to avoid listing files on every store operation
    private String cachedLatestFile = null;
    private int cachedFileRecordCount = 0;

    // Atomic sequence number for thread-safe increment
    private AtomicLong sequenceNumber = new AtomicLong(0);

    // Override ALL_FIELDS to include our custom configuration fields
    // This ensures that our custom fields are properly validated by Debezium
    public static final Field.Set ALL_FIELDS =
            Field.setOf(
                    Field.create(SOURCE_ID, "Unique source ID for schema history storage")
                            .withDescription(
                                    "A unique identifier for this source to avoid path conflicts between multiple sources")
                            .required(),
                    Field.create(
                                    MAX_RECORDS_PER_FILE_CONFIG,
                                    "Maximum number of records per schema history file")
                            .withDescription(
                                    "Maximum number of schema history records to store in a single file before creating a new file")
                            .withDefault(2048));

    @Override
    public void configure(
            Configuration config,
            HistoryRecordComparator comparator,
            SchemaHistoryListener listener,
            boolean useCatalogBeforeSchema) {
        super.configure(config, comparator, listener, useCatalogBeforeSchema);

Review comment (performance):
Can the maxRecordsPerFile be modified fo existing sources? It helps mitigate memory overhead caused by high memory consumption in cachedLatestFileRecords, e.g. when there're many sources.

Code after:
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class OpendalSchemaHistory extends AbstractFileBasedSchemaHistory {
    private static final Logger LOGGER = LoggerFactory.getLogger(OpendalSchemaHistory.class);
    private String sourceId = "";
    public static final String SOURCE_ID = "schema.history.internal.source.id";
    public static final String MAX_RECORDS_PER_FILE_CONFIG =
            "schema.history.internal.max.records.per.file";
    private int maxRecordsPerFile = 2048; // default records nums per file
    private String objectDir = "";

    private static final Pattern HISTORY_FILE_PATTERN =
            Pattern.compile("schema_history_(\\d+)\\.dat");

    // Cache the latest file information to avoid listing files on every store operation
    private String cachedLatestFile = null;
    // Cache the actual records from the latest file to avoid redundant getObject() calls
    private List<HistoryRecord> cachedLatestFileRecords = null;

    // Atomic sequence number for thread-safe increment
    private AtomicLong sequenceNumber = new AtomicLong(0);

    // Override ALL_FIELDS to include our custom configuration fields
    // This ensures that our custom fields are properly validated by Debezium
    public static final Field.Set ALL_FIELDS =
            Field.setOf(
                    Field.create(SOURCE_ID, "Unique source ID for schema history storage")
                            .withDescription(
                                    "A unique identifier for this source to avoid path conflicts between multiple sources")
                            .required(),
                    Field.create(
                                    MAX_RECORDS_PER_FILE_CONFIG,
                                    "Maximum number of records per schema history file")
                            .withDescription(
                                    "Maximum number of schema history records to store in a single file before creating a new file")
                            .withDefault(2048));

    @Override
    public void configure(
            Configuration config,
            HistoryRecordComparator comparator,
            SchemaHistoryListener listener,
            boolean useCatalogBeforeSchema) {