From 2c443ee5664c0307d7b7f50491fa6130a83faef5 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Fri, 17 Jul 2026 05:43:08 +0000
Subject: [PATCH] [#708] Fix OutOfMemory during replication initialize with JDBC backend (#711)
---
opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java | 156 +++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/Storage.java | 256 ++++++++++++++++++++----------------
opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/EncryptedTestCase.java | 3
opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/PluggableBackendImplTestCase.java | 4
4 files changed, 299 insertions(+), 120 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/Storage.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/Storage.java
index 2fea7f4..cc158fe 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/Storage.java
+++ b/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;
}
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/EncryptedTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/EncryptedTestCase.java
index 6b0279c..88cfbaa 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/EncryptedTestCase.java
+++ b/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");
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java
index efcc1a8..03316e2 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java
+++ b/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();
+ }
+ }
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/PluggableBackendImplTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/PluggableBackendImplTestCase.java
index 7568030..cae7c05 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/backends/pluggable/PluggableBackendImplTestCase.java
+++ b/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[]{})
);
--
Gitblit v1.10.0