OPENDJ-2580 File-based changelog growing and filling up the disk when using...

Fabio Pistolesi

05.31.2016 761ba8a31ca6ea11cd1965ab66c3bea4bcd5a01a

OPENDJ-2580 File-based changelog growing and filling up the disk when using modrate

Configuring a server as RS only should disable the change indexer. Change-log purging is based on it, but if it disabled, it should purge on purge-delay only.
Change from testing only the indexer being enabled to also checking there actually is at least a domain for which external change-log is enabled, so that indexer had data to work on and can return a purge CSN.

 opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/MultimasterReplication.java

@@ -22,7 +22,7 @@
 *
 *
 *      Copyright 2006-2010 Sun Microsystems, Inc.
 *      Portions Copyright 2011-2015 ForgeRock AS
 *      Portions Copyright 2011-2016 ForgeRock AS
 */
package org.opends.server.replication.plugin;

@@ -866,6 +866,31 @@
   */
  public static boolean isECLEnabledDomain(DN baseDN)
  {
    waitForStartup();
    // if state is STOPPING, then we need to return from this method
    final LDAPReplicationDomain domain = domains.get(baseDN);
    return domain != null && domain.isECLEnabled();
  }

  /**
   * Returns whether the external change-log contains data from at least a domain.
   * @return whether the external change-log contains data from at least a domain
   */
  public static boolean isECLEnabled()
  {
    waitForStartup();
    for (LDAPReplicationDomain domain : domains.values())
    {
      if (domain.isECLEnabled())
      {
        return true;
      }
    }
    return false;
  }

  private static void waitForStartup()
  {
    if (State.STARTING.equals(state.get()))
    {
      synchronized (state)
@@ -883,9 +908,6 @@
        }
      }
    }
    // if state is STOPPING, then we need to return from this method
    final LDAPReplicationDomain domain = domains.get(baseDN);
    return domain != null && domain.isECLEnabled();
  }

  /**

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServer.java

@@ -79,7 +79,7 @@
 * servers It is responsible for creating the replication server
 * replicationServerDomain and managing it
 */
public final class ReplicationServer
public class ReplicationServer
  implements ConfigurationChangeListener<ReplicationServerCfg>
{
  private String serverURL;
@@ -1338,6 +1338,15 @@
    return config.isComputeChangeNumber();
  }

  /**
   * Returns whether the external change-log contains data from at least a domain.
   * @return whether the external change-log contains data from at least a domain
   */
  public boolean isECLEnabled()
  {
    return MultimasterReplication.isECLEnabled();
  }

  /** {@inheritDoc} */
  @Override
  public String toString()

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/FileChangelogDB.java

@@ -904,7 +904,7 @@
          final CSN purgeCSN = new CSN(purgeTimestamp, 0, 0);
          final CSN oldestNotPurgedCSN;

          if (!replicationServer.isChangeNumberEnabled())
          if (!replicationServer.isChangeNumberEnabled() || !replicationServer.isECLEnabled())
          {
            oldestNotPurgedCSN = purgeCSN;
          }
@@ -1005,7 +1005,7 @@
    {
      final long nextPurgeTime = notPurgedCSN.getTime();
      final long currentPurgeTime = TimeThread.getTime() - purgeDelayInMillis;
      if (currentPurgeTime <= nextPurgeTime)
      if (currentPurgeTime < nextPurgeTime)
      {
        // sleep till the next CSN to purge,
        return nextPurgeTime - currentPurgeTime;

 opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/FileChangeNumberIndexDBTest.java

@@ -21,7 +21,7 @@
 * CDDL HEADER END
 *
 *
 *      Copyright 2014-2015 ForgeRock AS
 *      Copyright 2014-2016 ForgeRock AS
 */
package org.opends.server.replication.server.changelog.file;

@@ -29,6 +29,7 @@
import static org.opends.server.replication.server.changelog.file.FileReplicaDBTest.*;
import static org.testng.Assert.*;

import org.forgerock.opendj.config.server.ConfigException;
import org.forgerock.opendj.ldap.ByteString;
import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
@@ -78,7 +79,7 @@
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = newReplicationServer();
      replicationServer = newReplicationServer(false);
      final ChangelogDB changelogDB = replicationServer.getChangelogDB();
      changelogDB.setPurgeDelay(0);
      final FileChangeNumberIndexDB cnIndexDB = getCNIndexDB(replicationServer);
@@ -120,7 +121,7 @@
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = newReplicationServer();
      replicationServer = newReplicationServer(false);
      final ChangelogDB changelogDB = replicationServer.getChangelogDB();
      changelogDB.setPurgeDelay(0);
      final FileChangeNumberIndexDB cnIndexDB = getCNIndexDB(replicationServer);
@@ -150,7 +151,7 @@
    ReplicationServer replicationServer = null;
    try
    {
      replicationServer = newReplicationServer();
      replicationServer = newReplicationServer(true);
      final ChangelogDB changelogDB = replicationServer.getChangelogDB();
      changelogDB.setPurgeDelay(0); // disable purging

@@ -234,12 +235,31 @@
    assertEquals(oldest.getCSN(), newest.getCSN());
  }

  private ReplicationServer newReplicationServer() throws Exception
  private class FakeRS extends ReplicationServer
  {
    FakeRS(ReplServerFakeConfiguration cfg) throws ConfigException
    {
      super(cfg);
    }

    @Override
    public boolean isECLEnabled()
    {
      return true;
    }
  }

  private ReplicationServer newReplicationServer(boolean mock) throws Exception
  {
    TestCaseUtils.startServer();
    final int port = TestCaseUtils.findFreePort();
    final ReplServerFakeConfiguration cfg = new ReplServerFakeConfiguration(port, null, 0, 2, 0, 100, null);
    cfg.setComputeChangeNumber(true);
    if (mock)
    {
      FakeRS mockRS = new FakeRS(cfg);
      return mockRS;
    }
    return new ReplicationServer(cfg);
  }