bugfix: dsconfig on loggers trigger thread leaks

Jean-Noël Rouvignac

01.02.2015 7166491c42d43a48beac42bea796916019aee89e

bugfix: dsconfig on loggers trigger thread leaks

When applying configuration changes, the code was leaking AsynchronousTextWriters and ParallelTextWriters.
These two classes create threads in their constructors.
Leaking such objects end up creating way too many threads which in turn have a bad effect on the whole system.

Text*LogPublisher.java:
In applyConfigurationChange(), make sure:
- no [Asynchronous|Parallel]TextWriters are leaked
- the logic applying the changes only creates new objects when necessary
Added methods useAsyncWriter(), hasAsyncConfigChanged(), hasParallelConfigChanged().

 opendj-sdk/opendj-server-legacy/src/main/java/org/opends/server/loggers/TextAccessLogPublisher.java

@@ -118,30 +118,72 @@

        if (config.isAsynchronous())
        {
          // The asynchronous setting is being turned on.
          if (!(writer instanceof AsynchronousTextWriter))
          if (useAsyncWriter(config))
          {
            writer = newAsyncWriter(mfWriter, config);
            if (writer instanceof AsynchronousTextWriter)
            {
              if (hasAsyncConfigChanged(config))
              {
                // reinstantiate
                final AsynchronousTextWriter previousWriter = (AsynchronousTextWriter) writer;
                writer = newAsyncWriter(mfWriter, config);
                previousWriter.shutdown(false);
              }
            }
            else if (writer instanceof ParallelTextWriter)
            {
              // convert parallel to async
              final ParallelTextWriter previousWriter = (ParallelTextWriter) writer;
              writer = newAsyncWriter(mfWriter, config);
              previousWriter.shutdown(false);
            }
            else
            {
              // turn async text writer on
              writer = newAsyncWriter(mfWriter, config);
            }
          }
          if (!(writer instanceof ParallelTextWriter))
          else
          {
            writer = newParallelWriter(mfWriter, config);
            if (writer instanceof AsynchronousTextWriter)
            {
              // convert async to parallel
              final AsynchronousTextWriter previousWriter = (AsynchronousTextWriter) writer;
              writer = newParallelWriter(mfWriter, config);
              previousWriter.shutdown(false);
            }
            else if (writer instanceof ParallelTextWriter)
            {
              if (hasParallelConfigChanged(config))
              {
                // reinstantiate
                final ParallelTextWriter previousWriter = (ParallelTextWriter) writer;
                writer = newParallelWriter(mfWriter, config);
                previousWriter.shutdown(false);
              }
            }
            else
            {
              // turn parallel text writer on
              writer = newParallelWriter(mfWriter, config);
            }
          }
        }
        else
        {
          // The asynchronous setting is being turned off.
          if (writer instanceof AsynchronousTextWriter)
          {
            final AsynchronousTextWriter asyncWriter = (AsynchronousTextWriter) writer;
            writer = mfWriter;
            asyncWriter.shutdown(false);
          }
          if (writer instanceof ParallelTextWriter)
          {
            final ParallelTextWriter asyncWriter = (ParallelTextWriter) writer;
            // asynchronous is being turned off, remove parallel text writers.
            final ParallelTextWriter previousWriter = (ParallelTextWriter) writer;
            writer = mfWriter;
            asyncWriter.shutdown(false);
            previousWriter.shutdown(false);
          }
          else if (writer instanceof AsynchronousTextWriter)
          {
            // asynchronous is being turned off, remove async text writers.
            final AsynchronousTextWriter previousWriter = (AsynchronousTextWriter) writer;
            writer = mfWriter;
            previousWriter.shutdown(false);
          }
        }

@@ -234,7 +276,7 @@

      if (cfg.isAsynchronous())
      {
        if (cfg.getQueueSize() > 0)
        if (useAsyncWriter(cfg))
        {
          this.writer = newAsyncWriter(theWriter, cfg);
        }
@@ -269,6 +311,21 @@
    cfg.addFileBasedAccessChangeListener(this);
  }

  private boolean useAsyncWriter(FileBasedAccessLogPublisherCfg config)
  {
    return config.getQueueSize() > 0;
  }

  private boolean hasAsyncConfigChanged(FileBasedAccessLogPublisherCfg newConfig)
  {
    return hasParallelConfigChanged(newConfig) && cfg.getQueueSize() != newConfig.getQueueSize();
  }

  private boolean hasParallelConfigChanged(FileBasedAccessLogPublisherCfg newConfig)
  {
    return !cfg.dn().equals(newConfig.dn()) && cfg.isAutoFlush() != newConfig.isAutoFlush();
  }

  private AsynchronousTextWriter newAsyncWriter(MultifileTextWriter mfWriter, FileBasedAccessLogPublisherCfg config)
  {
    String name = "Asynchronous Text Writer for " + config.dn();

 opendj-sdk/opendj-server-legacy/src/main/java/org/opends/server/loggers/TextAuditLogPublisher.java

@@ -82,9 +82,19 @@

        if (config.isAsynchronous())
        {
          if (!(writer instanceof AsynchronousTextWriter))
          if (writer instanceof AsynchronousTextWriter)
          {
            // The asynchronous setting is being turned on.
            if (hasAsyncConfigChanged(config))
            {
              // reinstantiate
              final AsynchronousTextWriter previousWriter = (AsynchronousTextWriter) writer;
              writer = newAsyncWriter(mfWriter, config);
              previousWriter.shutdown(false);
            }
          }
          else
          {
            // turn async text writer on
            writer = newAsyncWriter(mfWriter, config);
          }
        }
@@ -92,10 +102,10 @@
        {
          if (writer instanceof AsynchronousTextWriter)
          {
            // The asynchronous setting is being turned off.
            AsynchronousTextWriter asyncWriter = (AsynchronousTextWriter) writer;
            // asynchronous is being turned off, remove async text writers.
            final AsynchronousTextWriter previousWriter = (AsynchronousTextWriter) writer;
            writer = mfWriter;
            asyncWriter.shutdown(false);
            previousWriter.shutdown(false);
          }
        }

@@ -150,6 +160,13 @@
    return getFileForPath(config.getLogFile());
  }

  private boolean hasAsyncConfigChanged(FileBasedAuditLogPublisherCfg newConfig)
  {
    return !cfg.dn().equals(newConfig.dn())
        && cfg.isAutoFlush() != newConfig.isAutoFlush()
        && cfg.getQueueSize() != newConfig.getQueueSize();
  }

  @Override
  protected void close0()
  {

 opendj-sdk/opendj-server-legacy/src/main/java/org/opends/server/loggers/TextDebugLogPublisher.java

@@ -232,9 +232,19 @@

        if (config.isAsynchronous())
        {
          if (!(writer instanceof AsynchronousTextWriter))
          if (writer instanceof AsynchronousTextWriter)
          {
            // The asynchronous setting is being turned on.
            if (hasAsyncConfigChanged(config))
            {
              // reinstantiate
              final AsynchronousTextWriter previousWriter = (AsynchronousTextWriter) writer;
              writer = newAsyncWriter(mfWriter, config);
              previousWriter.shutdown(false);
            }
          }
          else
          {
            // turn async text writer on
            writer = newAsyncWriter(mfWriter, config);
          }
        }
@@ -242,10 +252,10 @@
        {
          if (writer instanceof AsynchronousTextWriter)
          {
            // The asynchronous setting is being turned off.
            AsynchronousTextWriter asyncWriter = (AsynchronousTextWriter) writer;
            // asynchronous is being turned off, remove async text writers.
            final AsynchronousTextWriter previousWriter = (AsynchronousTextWriter) writer;
            writer = mfWriter;
            asyncWriter.shutdown(false);
            previousWriter.shutdown(false);
          }
        }

@@ -306,6 +316,13 @@
    return getFileForPath(config.getLogFile());
  }

  private boolean hasAsyncConfigChanged(FileBasedDebugLogPublisherCfg newConfig)
  {
    return !currentConfig.dn().equals(newConfig.dn())
        && currentConfig.isAutoFlush() != newConfig.isAutoFlush()
        && currentConfig.getQueueSize() != newConfig.getQueueSize();
  }

  private TraceSettings getDefaultSettings(FileBasedDebugLogPublisherCfg config)
  {
    return new TraceSettings(

 opendj-sdk/opendj-server-legacy/src/main/java/org/opends/server/loggers/TextErrorLogPublisher.java

@@ -352,9 +352,19 @@

        if (config.isAsynchronous())
        {
          if (!(writer instanceof AsynchronousTextWriter))
          if (writer instanceof AsynchronousTextWriter)
          {
            // The asynchronous setting is being turned on.
            if (hasAsyncConfigChanged(config))
            {
              // reinstantiate
              final AsynchronousTextWriter previousWriter = (AsynchronousTextWriter) writer;
              writer = newAsyncWriter(mfWriter, config);
              previousWriter.shutdown(false);
            }
          }
          else
          {
            // turn async text writer on
            writer = newAsyncWriter(mfWriter, config);
          }
        }
@@ -362,15 +372,14 @@
        {
          if (writer instanceof AsynchronousTextWriter)
          {
            // The asynchronous setting is being turned off.
            AsynchronousTextWriter asyncWriter = (AsynchronousTextWriter) writer;
            // asynchronous is being turned off, remove async text writers.
            final AsynchronousTextWriter previousWriter = (AsynchronousTextWriter) writer;
            writer = mfWriter;
            asyncWriter.shutdown(false);
            previousWriter.shutdown(false);
          }
        }

        if (currentConfig.isAsynchronous()
            && config.isAsynchronous()
        if (currentConfig.isAsynchronous() && config.isAsynchronous()
            && currentConfig.getQueueSize() != config.getQueueSize())
        {
          ccr.setAdminActionRequired(true);
@@ -421,6 +430,13 @@
    return getFileForPath(config.getLogFile());
  }

  private boolean hasAsyncConfigChanged(FileBasedErrorLogPublisherCfg newConfig)
  {
    return !currentConfig.dn().equals(newConfig.dn())
        && currentConfig.isAutoFlush() != newConfig.isAutoFlush()
        && currentConfig.getQueueSize() != newConfig.getQueueSize();
  }

  private AsynchronousTextWriter newAsyncWriter(MultifileTextWriter mfWriter, FileBasedErrorLogPublisherCfg config)
  {
    String name = "Asynchronous Text Writer for " + config.dn();

 opendj-sdk/opendj-server-legacy/src/main/java/org/opends/server/loggers/TextHTTPAccessLogPublisher.java

@@ -140,34 +140,72 @@

        if (config.isAsynchronous())
        {
          if (!(writer instanceof AsynchronousTextWriter))
          if (useAsyncWriter(config))
          {
            // The asynchronous setting is being turned on.
            final AsynchronousTextWriter asyncWriter = newAsyncWriter(mfWriter, config);
            writer = asyncWriter;
            if (writer instanceof AsynchronousTextWriter)
            {
              if (hasAsyncConfigChanged(config))
              {
                // reinstantiate
                final AsynchronousTextWriter previousWriter = (AsynchronousTextWriter) writer;
                writer = newAsyncWriter(mfWriter, config);
                previousWriter.shutdown(false);
              }
            }
            else if (writer instanceof ParallelTextWriter)
            {
              // convert parallel to async
              final ParallelTextWriter previousWriter = (ParallelTextWriter) writer;
              writer = newAsyncWriter(mfWriter, config);
              previousWriter.shutdown(false);
            }
            else
            {
              // turn async text writer on
              writer = newAsyncWriter(mfWriter, config);
            }
          }
          if (!(writer instanceof ParallelTextWriter))
          else
          {
            // The asynchronous setting is being turned on.
            final ParallelTextWriter asyncWriter = newParallelWriter(mfWriter, config);
            writer = asyncWriter;
            if (writer instanceof AsynchronousTextWriter)
            {
              // convert async to parallel
              final AsynchronousTextWriter previousWriter = (AsynchronousTextWriter) writer;
              writer = newParallelWriter(mfWriter, config);
              previousWriter.shutdown(false);
            }
            else if (writer instanceof ParallelTextWriter)
            {
              if (hasParallelConfigChanged(config))
              {
                // reinstantiate
                final ParallelTextWriter previousWriter = (ParallelTextWriter) writer;
                writer = newParallelWriter(mfWriter, config);
                previousWriter.shutdown(false);
              }
            }
            else
            {
              // turn parallel text writer on
              writer = newParallelWriter(mfWriter, config);
            }
          }
        }
        else
        {
          if (writer instanceof AsynchronousTextWriter)
          {
            // The asynchronous setting is being turned off.
            final AsynchronousTextWriter asyncWriter = (AsynchronousTextWriter) writer;
            writer = mfWriter;
            asyncWriter.shutdown(false);
          }
          if (writer instanceof ParallelTextWriter)
          {
            // The asynchronous setting is being turned off.
            final ParallelTextWriter asyncWriter = (ParallelTextWriter) writer;
            // asynchronous is being turned off, remove parallel text writers.
            final ParallelTextWriter previousWriter = (ParallelTextWriter) writer;
            writer = mfWriter;
            asyncWriter.shutdown(false);
            previousWriter.shutdown(false);
          }
          else if (writer instanceof AsynchronousTextWriter)
          {
            // asynchronous is being turned off, remove async text writers.
            final AsynchronousTextWriter previousWriter = (AsynchronousTextWriter) writer;
            writer = mfWriter;
            previousWriter.shutdown(false);
          }
        }

@@ -237,10 +275,19 @@
    return getFileForPath(config.getLogFile());
  }

  private ParallelTextWriter newParallelWriter(MultifileTextWriter mfWriter, FileBasedHTTPAccessLogPublisherCfg config)
  private boolean useAsyncWriter(FileBasedHTTPAccessLogPublisherCfg config)
  {
    String name = "Parallel Text Writer for " + config.dn();
    return new ParallelTextWriter(name, config.isAutoFlush(), mfWriter);
    return config.getQueueSize() > 0;
  }

  private boolean hasAsyncConfigChanged(FileBasedHTTPAccessLogPublisherCfg newConfig)
  {
    return hasParallelConfigChanged(newConfig) && cfg.getQueueSize() != newConfig.getQueueSize();
  }

  private boolean hasParallelConfigChanged(FileBasedHTTPAccessLogPublisherCfg newConfig)
  {
    return !cfg.dn().equals(newConfig.dn()) && cfg.isAutoFlush() != newConfig.isAutoFlush();
  }

  private AsynchronousTextWriter newAsyncWriter(MultifileTextWriter mfWriter, FileBasedHTTPAccessLogPublisherCfg config)
@@ -249,6 +296,12 @@
    return new AsynchronousTextWriter(name, config.getQueueSize(), config.isAutoFlush(), mfWriter);
  }

  private ParallelTextWriter newParallelWriter(MultifileTextWriter mfWriter, FileBasedHTTPAccessLogPublisherCfg config)
  {
    String name = "Parallel Text Writer for " + config.dn();
    return new ParallelTextWriter(name, config.isAutoFlush(), mfWriter);
  }

  private List<String> extractFieldsOrder(String logFormat)
  {
    // there will always be at least one field value due to the regexp
@@ -333,7 +386,7 @@

      if (cfg.isAsynchronous())
      {
        if (cfg.getQueueSize() > 0)
        if (useAsyncWriter(cfg))
        {
          this.writer = newAsyncWriter(theWriter, cfg);
        }