[#708] Fix OutOfMemory during replication initialize with JDBC backend (#711)

Valery Kharseko

yesterday 2c443ee5664c0307d7b7f50491fa6130a83faef5

[#708] Fix OutOfMemory during replication initialize with JDBC backend (#711)

 opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/Storage.java

@@ -11,7 +11,7 @@
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2024-2025 3A Systems, LLC.
 * Copyright 2024-2026 3A Systems, LLC.
 */
package org.opends.server.backends.jdbc;

@@ -282,7 +282,7 @@
            if (((CachedConnection) con).parent.getClass().getName().contains("oracle")) {
                return "h char(128),k raw(2000),v blob,primary key(h,k)";
            }else if (((CachedConnection) con).parent.getClass().getName().contains("mysql")) {
                return "h char(128),k tinyblob,v longblob,primary key(h(128),k(255))";
                return "h char(128),k varbinary(255),v longblob,primary key(h,k)";
            }else if (((CachedConnection) con).parent.getClass().getName().contains("microsoft")) {
                return "h char(128),k varbinary(max),v image,primary key(h)";
            }
@@ -291,15 +291,50 @@

        @Override
        public void openTree(TreeName treeName, boolean createOnDemand) {
            if (createOnDemand && !isExistsTable(treeName)) {
                try (final PreparedStatement statement=con.prepareStatement("create table "+getTableName(treeName)+" ("+getTableDialect()+")")){
                    execute(statement);
                    con.commit();
                }catch (SQLException e) {
                    throw new StorageRuntimeException(e);
            if (createOnDemand) {
                if (!isExistsTable(treeName)) {
                    try (final PreparedStatement statement=con.prepareStatement("create table "+getTableName(treeName)+" ("+getTableDialect()+")")){
                        execute(statement);
                        con.commit();
                    }catch (SQLException e) {
                        throw new StorageRuntimeException(e);
                    }
                }
                // CursorImpl iterates with "where k>? order by k" batches: primary key (h,k) cannot serve them
                final String driverName=((CachedConnection) con).parent.getClass().getName();
                final String tableName=getTableName(treeName);
                if (driverName.contains("postgres")) {
                    try (final PreparedStatement statement=con.prepareStatement("create index if not exists k_"+tableName.substring("opendj_".length())+" on "+tableName+" (k)")){
                        execute(statement);
                        con.commit();
                    }catch (SQLException e) {
                        throw new StorageRuntimeException(e);
                    }
                }else if (driverName.contains("mysql")) {
                    try {
                        if (!isExistsIndex(tableName,"k_"+tableName.substring("opendj_".length()))) { // mysql has no "create index if not exists"
                            try (final PreparedStatement statement=con.prepareStatement("create index k_"+tableName.substring("opendj_".length())+" on "+tableName+" (k)")){
                                execute(statement);
                                con.commit();
                            }
                        }
                    }catch (SQLException e) {
                        throw new StorageRuntimeException(e);
                    }
                }
            }
        }

        boolean isExistsIndex(String tableName, String indexName) throws SQLException {
            try (final ResultSet rs = con.getMetaData().getIndexInfo(null, null, tableName, false, false)) {
                while (rs.next()) {
                    if (indexName.equalsIgnoreCase(rs.getString("INDEX_NAME"))) {
                        return true;
                    }
                }
            }
            return false;
        }
        
        public void clearTree(TreeName treeName) {
            try (final PreparedStatement statement=con.prepareStatement("delete from "+getTableName(treeName))){
@@ -414,74 +449,102 @@
        }
    }
    
    // Iterates in batches of "fetchsize" records via keyset pagination ("where k>? order by k limit n"):
    // scrollable ResultSet is not an option, the postgres/mysql drivers materialize it entirely in memory.
    private final class CursorImpl implements Cursor<ByteString, ByteString> {
        final TreeName treeName;

        final PreparedStatement statement;
        final ResultSet rc;
        final Connection con;
        final String tableName;
        final boolean isReadOnly;
        final int batchSize=Math.max(1,Integer.getInteger("org.openidentityplatform.opendj.jdbc.fetchsize",1000));
        final String limitClause;

        final ArrayDeque<byte[][]> buffer=new ArrayDeque<>();
        byte[] currentKeyDb;
        ByteString currentKey;
        ByteString currentValue;
        boolean defined;

        public CursorImpl(boolean isReadOnly, Connection con, TreeName treeName) {
            this.treeName=treeName;
            this.isReadOnly=isReadOnly;
            try {
                statement=con.prepareStatement("select h,k,v from "+getTableName(treeName)+" order by k",
                    isReadOnly?ResultSet.TYPE_SCROLL_INSENSITIVE:ResultSet.TYPE_SCROLL_SENSITIVE,
                    isReadOnly?ResultSet.CONCUR_READ_ONLY:ResultSet.CONCUR_UPDATABLE);
                rc=executeResultSet(statement);
            this.con=con;
            this.tableName=getTableName(treeName);
            this.limitClause=((CachedConnection)con).parent.getClass().getName().contains("mysql")
                ? " limit ?,?" : " offset ? rows fetch next ? rows only";
        }

        boolean fetchBatch(String condition, byte[] dbKey, long offset, boolean descending, int limit) {
            buffer.clear();
            try (final PreparedStatement statement=con.prepareStatement("select k,v from "+tableName
                    +(condition!=null?" where k"+condition+"?":"")
                    +" order by k"+(descending?" desc":"")+limitClause)){
                int i=1;
                if (condition!=null) {
                    statement.setBytes(i++,dbKey);
                }
                statement.setLong(i++,offset);
                statement.setLong(i,limit);
                try(final ResultSet rc=executeResultSet(statement)) {
                    while (rc.next()) {
                        buffer.add(new byte[][]{rc.getBytes(1),rc.getBytes(2)});
                    }
                }
            }catch (SQLException e) {
                throw new StorageRuntimeException(e);
            }
            return !buffer.isEmpty();
        }

        void advanceFromBuffer() {
            final byte[][] row=buffer.poll();
            currentKeyDb=row[0];
            currentKey=ByteString.wrap(db2real(row[0]));
            currentValue=ByteString.wrap(row[1]);
            defined=true;
        }

        @Override
        public boolean next() {
            try {
                return rc.next();
            }catch (SQLException e) {
                throw new StorageRuntimeException(e);
            if (buffer.isEmpty() && !fetchBatch(currentKeyDb==null?null:">",currentKeyDb,0,false,batchSize)) {
                defined=false;
                return false;
            }
            advanceFromBuffer();
            return true;
        }

        @Override
        public boolean isDefined() {
            try{
                return rc.getRow()>0;
            }catch (SQLException e) {
                throw new StorageRuntimeException(e);
            }
            return defined;
        }

        @Override
        public ByteString getKey() throws NoSuchElementException {
            if (!isDefined()) {
            if (!defined) {
                throw new NoSuchElementException();
            }
            try{
                return ByteString.wrap(db2real(rc.getBytes("k")));
            }catch (SQLException e) {
                throw new StorageRuntimeException(e);
            }
            return currentKey;
        }

        @Override
        public ByteString getValue() throws NoSuchElementException {
            if (!isDefined()) {
            if (!defined) {
                throw new NoSuchElementException();
            }
            try{
                return ByteString.wrap(rc.getBytes("v"));
            }catch (SQLException e) {
                throw new StorageRuntimeException(e);
            }
            return currentValue;
        }

        @Override
        public void delete() throws NoSuchElementException, UnsupportedOperationException {
            if (!isDefined()) {
            if (!defined) {
                throw new NoSuchElementException();
            }
            try{
                rc.deleteRow();
            if (isReadOnly) {
                throw new UnsupportedOperationException();
            }
            try (final PreparedStatement statement=con.prepareStatement("delete from "+tableName+" where h=? and k=?")){
                statement.setString(1,key2hash.get(ByteBuffer.wrap(db2real(currentKeyDb))));
                statement.setBytes(2,currentKeyDb);
                execute(statement);
            }catch (SQLException e) {
                throw new StorageRuntimeException(e);
            }
@@ -489,97 +552,60 @@

        @Override
        public void close() {
            try{
                rc.close();
                statement.close();
            }catch (SQLException e) {
                throw new StorageRuntimeException(e);
            }
            buffer.clear();
            defined=false;
        }


        @Override
        public boolean positionToKeyOrNext(ByteSequence key) {
            if (!isDefined() || key.compareTo(getKey())<0) { //restart iterator
                try{
                    rc.first();
                }catch (SQLException e) {
                    throw new StorageRuntimeException(e);
                }
            }
            try{
                if (!isDefined()){
                    return false;
                }
                do {
                    if (key.compareTo(getKey())<=0) {
                        return true;
                    }
                }while(rc.next());
            }catch (SQLException e) {
                throw new StorageRuntimeException(e);
            }
            return false;
        }
		
        @Override
        public boolean positionToKey(ByteSequence key) {
            if (!isDefined() || key.compareTo(getKey())<0) {  //restart iterator
                try{
                    rc.first();
                }catch (SQLException e) {
                    throw new StorageRuntimeException(e);
                }
            }
            if (!isDefined()){
                return false;
            }
            if (isDefined() && key.compareTo(getKey())==0) {
            if (fetchBatch(">=",real2db(key.toByteArray()),0,false,batchSize)) {
                advanceFromBuffer();
                return true;
            }
            try{
                do {
                    if (key.compareTo(getKey())==0) {
                        return true;
                    }
                }while(rc.next());
            }catch (SQLException e) {
                throw new StorageRuntimeException(e);
            }
            defined=false;
            return false;
        }

		
        @Override
        public boolean positionToLastKey() {
            try{
                return rc.last();
        public boolean positionToKey(ByteSequence key) {
            final byte[] real=key.toByteArray();
            try (final PreparedStatement statement=con.prepareStatement("select v from "+tableName+" where h=? and k=?")){
                statement.setString(1,key2hash.get(ByteBuffer.wrap(real)));
                statement.setBytes(2,real2db(real));
                try(final ResultSet rc=executeResultSet(statement)) {
                    if (rc.next()) {
                        buffer.clear();
                        currentKeyDb=real2db(real);
                        currentKey=ByteString.wrap(real);
                        currentValue=ByteString.wrap(rc.getBytes("v"));
                        defined=true;
                        return true;
                    }
                }
            }catch (SQLException e) {
                throw new StorageRuntimeException(e);
            }
            defined=false;
            return false;
        }

        @Override
        public boolean positionToLastKey() {
            if (fetchBatch(null,null,0,true,1)) {
                advanceFromBuffer();
                return true;
            }
            defined=false;
            return false;
        }

        @Override
        public boolean positionToIndex(int index) {
            try{
                rc.first();
            }catch (SQLException e) {
                throw new StorageRuntimeException(e);
            if (index>=0 && fetchBatch(null,null,index,false,batchSize)) {
                advanceFromBuffer();
                return true;
            }
            if (!isDefined()){
                return false;
            }
            int ct=0;
            try{
                do {
                    if (ct==index) {
                        return true;
                    }
                    ct++;
                }while(rc.next());
            }catch (SQLException e) {
                throw new StorageRuntimeException(e);
            }
            defined=false;
            return false;
        }
    }

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/EncryptedTestCase.java

@@ -11,7 +11,7 @@
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2023-2024 3A Systems, LLC.
 * Copyright 2023-2026 3A Systems, LLC.
 */
package org.opends.server.backends.jdbc;

@@ -46,6 +46,7 @@
            container.start();
        }
        try(Connection con= DriverManager.getConnection(createBackendCfg().getDBDirectory())){
            TestCase.dropStaleTrees(con);
        } catch (Exception e) {
            throw new SkipException("run before test: docker run --rm -it -p 5432:5432 -e POSTGRES_DB=database_name -e POSTGRES_PASSWORD=password --name postgres postgres");
        }

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java

@@ -11,23 +11,43 @@
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2024-2025 3A Systems, LLC.
 * Copyright 2024-2026 3A Systems, LLC.
 */
package org.opends.server.backends.jdbc;

import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
import org.opends.server.backends.pluggable.PluggableBackendImplTestCase;
import org.opends.server.backends.pluggable.spi.AccessMode;
import org.opends.server.backends.pluggable.spi.Cursor;
import org.opends.server.backends.pluggable.spi.ReadOperation;
import org.opends.server.backends.pluggable.spi.ReadableTransaction;
import org.opends.server.backends.pluggable.spi.TreeName;
import org.opends.server.backends.pluggable.spi.WriteOperation;
import org.opends.server.backends.pluggable.spi.WriteableTransaction;
import org.testcontainers.DockerClientFactory;
import org.testcontainers.containers.JdbcDatabaseContainer;
import org.testng.SkipException;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.NoSuchElementException;

import static org.forgerock.opendj.config.ConfigurationMock.mockCfg;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;



@@ -50,14 +70,35 @@
                throw new SkipException(getContainerDockerCommand());
            }
        }
        try(Connection ignored = DriverManager.getConnection(createBackendCfg().getDBDirectory())){

        try(Connection con = DriverManager.getConnection(createBackendCfg().getDBDirectory())){
            dropStaleTrees(con);
        } catch (Exception e) {
            throw new SkipException(getContainerDockerCommand());
        }
        super.setUp();
    }

    /**
     * Backend test classes sharing one database map the same tree names to the same tables,
     * so a previous run may leave trees behind — including entries encrypted with a lost cipher key.
     */
    static void dropStaleTrees(Connection con) throws SQLException {
        final List<String> stale = new ArrayList<>();
        try (final ResultSet rs = con.getMetaData().getTables(null, null, null, new String[]{"TABLE"})) {
            while (rs.next()) {
                final String name = rs.getString("TABLE_NAME");
                if (name.toLowerCase().startsWith("opendj_")) {
                    stale.add(name);
                }
            }
        }
        try (final Statement st = con.createStatement()) {
            for (final String name : stale) {
                st.execute("drop table " + name);
            }
        }
    }

    @Override
    protected Backend createBackend() {
        return new Backend();
@@ -87,4 +128,113 @@
    protected abstract String getBackendId();

    protected abstract String getJdbcUrl();

    private static ByteString key(int i) {
        return ByteString.valueOfUtf8(String.format("key%02d", i));
    }

    private static ByteString value(int i) {
        return ByteString.valueOfUtf8("value" + i);
    }

    /** Cursor operations must keep working when the tree spans several "fetchsize" batches. */
    @Test
    public void testCursorCrossesFetchSizeBatches() throws Exception {
        System.setProperty("org.openidentityplatform.opendj.jdbc.fetchsize", "2");
        final Storage storage = new Storage(createBackendCfg(), null);
        final TreeName tree = new TreeName("testCursorBatch", "tree");
        try {
            storage.open(AccessMode.READ_WRITE);
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    txn.openTree(tree, true);
                    for (int i = 0; i < 7; i++) {
                        txn.put(tree, key(i), value(i));
                    }
                }
            });
            storage.read(new ReadOperation<Void>() {
                @Override
                public Void run(ReadableTransaction txn) throws Exception {
                    try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
                        for (int i = 0; i < 7; i++) {
                            assertTrue(cursor.next(), "next() at " + i);
                            assertEquals(cursor.getKey(), key(i));
                            assertEquals(cursor.getValue(), value(i));
                        }
                        assertFalse(cursor.next());
                        assertFalse(cursor.isDefined());
                        try {
                            cursor.getKey();
                            fail("getKey() on undefined cursor must fail");
                        } catch (NoSuchElementException expected) {}

                        assertTrue(cursor.positionToKeyOrNext(key(3)));
                        assertEquals(cursor.getKey(), key(3));
                        assertTrue(cursor.positionToKeyOrNext(ByteString.valueOfUtf8("key031")));
                        assertEquals(cursor.getKey(), key(4));
                        assertTrue(cursor.next());
                        assertEquals(cursor.getKey(), key(5));
                        assertTrue(cursor.next());
                        assertEquals(cursor.getKey(), key(6));
                        assertFalse(cursor.next());
                        assertFalse(cursor.positionToKeyOrNext(ByteString.valueOfUtf8("z")));
                        assertFalse(cursor.isDefined());

                        assertTrue(cursor.positionToKey(key(5)));
                        assertEquals(cursor.getValue(), value(5));
                        assertTrue(cursor.next());
                        assertEquals(cursor.getKey(), key(6));
                        assertFalse(cursor.positionToKey(ByteString.valueOfUtf8("key99")));
                        assertFalse(cursor.isDefined());

                        assertTrue(cursor.positionToIndex(0));
                        assertEquals(cursor.getKey(), key(0));
                        assertTrue(cursor.positionToIndex(5));
                        assertEquals(cursor.getKey(), key(5));
                        assertTrue(cursor.next());
                        assertEquals(cursor.getKey(), key(6));
                        assertFalse(cursor.positionToIndex(7));
                        assertFalse(cursor.positionToIndex(-1));

                        assertTrue(cursor.positionToLastKey());
                        assertEquals(cursor.getKey(), key(6));
                        assertFalse(cursor.next());

                        assertTrue(cursor.positionToKey(key(0)));
                        try {
                            cursor.delete();
                            fail("delete() on read-only cursor must fail");
                        } catch (UnsupportedOperationException expected) {}
                    }
                    return null;
                }
            });
            storage.write(new WriteOperation() {
                @Override
                public void run(WriteableTransaction txn) throws Exception {
                    try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
                        assertTrue(cursor.positionToKey(key(3)));
                        cursor.delete();
                        assertTrue(cursor.next());
                        assertEquals(cursor.getKey(), key(4));
                    }
                    assertNull(txn.read(tree, key(3)));
                    assertEquals(txn.getRecordCount(tree), 6);
                }
            });
        } finally {
            System.clearProperty("org.openidentityplatform.opendj.jdbc.fetchsize");
            try {
                storage.write(new WriteOperation() {
                    @Override
                    public void run(WriteableTransaction txn) throws Exception {
                        txn.deleteTree(tree);
                    }
                });
            } catch (Exception ignored) {}
            storage.close();
        }
    }
}

 opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/PluggableBackendImplTestCase.java

@@ -1217,7 +1217,9 @@
              "objectClasses: ( 16.32.256.1.1.1.12.4.6 NAME 'examplePerson' DESC 'Extension to person' SUP inetOrgPerson STRUCTURAL MUST ( exampleIdentifier ) MAY ( exampleEmails $ userAccountControl ) X-SCHEMA-FILE '999-user.ldif' )",
              ""
            );
    assertEquals(resultCode, 0);
    // 20 = attributeOrValueExists: another backend test class already added these definitions
    // to the schema of the shared in-JVM server
    assertTrue(resultCode == 0 || resultCode == 20, "unexpected result code " + resultCode);
    TestCaseUtils.addEntries(
            Resources.readLines(Resources.getResource("issue496.ldif"), StandardCharsets.UTF_8).toArray(new String[]{})
    );