public static synchronized void configure(DataflowWorkerLoggingOptions options) {
    if (!initialized) {
      throw new RuntimeException("configure() called before initialize()");
    }

    // For compatibility reason, we do not call SdkHarnessOptions.getConfiguredLoggerFromOptions
    // to config the logging for legacy worker, instead replicate the config steps used for
    // DataflowWorkerLoggingOptions for default log level and log level overrides.
    SdkHarnessOptions harnessOptions = options.as(SdkHarnessOptions.class);
    boolean usedDeprecated = false;

    // default value for both DefaultSdkHarnessLogLevel and DefaultWorkerLogLevel are INFO
    Level overrideLevel = getJulLevel(harnessOptions.getDefaultSdkHarnessLogLevel());
    if (options.getDefaultWorkerLogLevel() != null && options.getDefaultWorkerLogLevel() != INFO) {
      overrideLevel = getJulLevel(options.getDefaultWorkerLogLevel());
      usedDeprecated = true;
    }
    LogManager.getLogManager().getLogger(ROOT_LOGGER_NAME).setLevel(overrideLevel);

    if (options.getWorkerLogLevelOverrides() != null) {
      for (Map.Entry<String, DataflowWorkerLoggingOptions.Level> loggerOverride :
          options.getWorkerLogLevelOverrides().entrySet()) {
        Logger logger = Logger.getLogger(loggerOverride.getKey());
        logger.setLevel(getJulLevel(loggerOverride.getValue()));
        configuredLoggers.add(logger);
      }
      usedDeprecated = true;
    } else if (harnessOptions.getSdkHarnessLogLevelOverrides() != null) {
      for (Map.Entry<String, SdkHarnessOptions.LogLevel> loggerOverride :
          harnessOptions.getSdkHarnessLogLevelOverrides().entrySet()) {
        Logger logger = Logger.getLogger(loggerOverride.getKey());
        logger.setLevel(getJulLevel(loggerOverride.getValue()));
        configuredLoggers.add(logger);
      }
    }

    // If the options specify a level for messages logged to System.out/err, we need to reconfigure
    // the corresponding stream adapter.
    if (options.getWorkerSystemOutMessageLevel() != null) {
      System.out.close();
      System.setOut(
          JulHandlerPrintStreamAdapterFactory.create(
              loggingHandler,
              SYSTEM_OUT_LOG_NAME,
              getJulLevel(options.getWorkerSystemOutMessageLevel()),
              Charset.defaultCharset()));
    }

    if (options.getWorkerSystemErrMessageLevel() != null) {
      System.err.close();
      System.setErr(
          JulHandlerPrintStreamAdapterFactory.create(
              loggingHandler,
              SYSTEM_ERR_LOG_NAME,
              getJulLevel(options.getWorkerSystemErrMessageLevel()),
              Charset.defaultCharset()));
    }

    if (usedDeprecated) {
      LOG.warn(
          "Deprecated DataflowWorkerLoggingOptions are used for log level settings."
              + "Consider using options defined in SdkHarnessOptions for forward compatibility.");
    }
  }
@Test
  public void testStrictGlobalFilterAndRelaxedOverride() throws IOException {
    DataflowWorkerLoggingOptions options =
        PipelineOptionsFactory.as(DataflowWorkerLoggingOptions.class);
    options.setDefaultWorkerLogLevel(DataflowWorkerLoggingOptions.Level.ERROR);
    options.setWorkerLogLevelOverrides(
        new WorkerLogLevelOverrides()
            .addOverrideForName("A", DataflowWorkerLoggingOptions.Level.INFO));

    DataflowWorkerLoggingInitializer.configure(options);
    LogManager.getLogManager().getLogger("A").info("foobar");

    verifyLogOutput("foobar");
  }