bugfix: dsconfig on loggers trigger thread leaks

Jean-Noël Rouvignac

01.02.2015 7166491c42d43a48beac42bea796916019aee89e

 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);
        }