[Dataflow Java] Fix incorrect warning log when specifying logger override. (#39034)

Author: scwhittle

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingInitializer.java

@@ -217,17 +217,16 @@ private static void applyOverridesToLogger(
       boolean directLoggingEnabled) {
     if (!directLoggingEnabled) {
       logger.setLevel(overrides.disk);
-      if (overrides.disk.intValue() != Level.OFF.intValue()) {
+      if (overrides.direct.intValue() != Level.OFF.intValue()) {
         LOG.warn(
-            "Ignoring the disk logging level override for {} because --defaultWorkerDirectLoggerLevel was OFF.",
+            "Ignoring the direct logging level override for {} because --defaultWorkerDirectLoggerLevel was OFF.",
             logger.getName());
       }
       return;
     }
 
     // Configure the logger to accept logs of the lower value. The log will be sent directly based
-    // upon the default
-    // nonDirectLogLevel or the provided hint.
+    // upon the default nonDirectLogLevel or the provided hint.
     if (overrides.direct.intValue() < overrides.disk.intValue()) {
       logger.setLevel(overrides.direct);
       if (overrides.disk.intValue() != defaultNonDirectLogLevel.intValue()) {
@@ -245,9 +244,8 @@ private static void applyOverridesToLogger(
       logger.setLevel(overrides.disk);
       if (overrides.disk.intValue() < defaultNonDirectLogLevel.intValue()) {
         // We need to provide the hint as otherwise the default would be used and send the log
-        // directly which is not
-        // desired. We don't bother if the threshold was increased as the right decision would still
-        // be made.
+        // directly which is not desired. We don't bother if the threshold was increased as the
+        // right decision would still be made.
         logger.setResourceBundle(
             DataflowWorkerLoggingHandler.resourceBundleForNonDirectLogLevelHint(overrides.disk));
       }

runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingInitializerTest.java

@@ -134,7 +134,7 @@ public void testWithSdkHarnessConfigurationOverride() {
   }
 
   @Test
-  public void testWithWorkerCustomLogLevels() {
+  public void testWithWorkerCustomLogLevels() throws IOException {
     DataflowWorkerLoggingOptions options =
         PipelineOptionsFactory.as(DataflowWorkerLoggingOptions.class);
     options.setWorkerLogLevelOverrides(
@@ -153,6 +153,9 @@ public void testWithWorkerCustomLogLevels() {
     assertEquals(Level.SEVERE, bLogger.getLevel());
     assertEquals(0, bLogger.getHandlers().length);
     assertTrue(aLogger.getUseParentHandlers());
+
+    List<String> logLines = retrieveLogLines();
+    assertThat(logLines, not(hasItem(containsString("Ignoring the direct logging level"))));
   }
 
   @Test
@@ -176,7 +179,7 @@ public void testWithDirectLogging() {
   }
 
   @Test
-  public void testWithSdkHarnessCustomLogLevels() {
+  public void testWithSdkHarnessCustomLogLevels() throws IOException {
     SdkHarnessOptions options = PipelineOptionsFactory.as(SdkHarnessOptions.class);
     options.setSdkHarnessLogLevelOverrides(
         new SdkHarnessLogLevelOverrides()
@@ -194,6 +197,43 @@ public void testWithSdkHarnessCustomLogLevels() {
     assertEquals(Level.SEVERE, bLogger.getLevel());
     assertEquals(0, bLogger.getHandlers().length);
     assertTrue(aLogger.getUseParentHandlers());
+
+    List<String> logLines = retrieveLogLines();
+    assertThat(logLines, not(hasItem(containsString("Ignoring the direct logging level"))));
+  }
+
+  @Test
+  public void testWithSdkHarnessCustomLogLevels_DirectSpecifiedNotEnabled() throws IOException {
+    SdkHarnessOptions options = PipelineOptionsFactory.as(SdkHarnessOptions.class);
+    options.setSdkHarnessLogLevelOverrides(
+        new SdkHarnessLogLevelOverrides()
+            .addOverrideForName("B", SdkHarnessOptions.LogLevel.DEBUG)
+            .addOverrideForName("C", SdkHarnessOptions.LogLevel.ERROR));
+    DataflowWorkerLoggingOptions loggingOptions = options.as(DataflowWorkerLoggingOptions.class);
+    loggingOptions.setWorkerDirectLogLevelOverrides(
+        new WorkerLogLevelOverrides()
+            .addOverrideForName("A", DataflowWorkerLoggingOptions.Level.TRACE)
+            .addOverrideForName("C", DataflowWorkerLoggingOptions.Level.TRACE));
+
+    DataflowWorkerLoggingInitializer.configure(options.as(DataflowWorkerLoggingOptions.class));
+
+    Logger aLogger = LogManager.getLogManager().getLogger("A");
+    assertEquals(0, aLogger.getHandlers().length);
+    assertEquals(Level.INFO, aLogger.getLevel());
+    assertTrue(aLogger.getUseParentHandlers());
+
+    Logger bLogger = LogManager.getLogManager().getLogger("B");
+    assertEquals(Level.FINE, bLogger.getLevel());
+    assertEquals(0, bLogger.getHandlers().length);
+    assertTrue(aLogger.getUseParentHandlers());
+
+    Logger cLogger = LogManager.getLogManager().getLogger("C");
+    assertEquals(Level.SEVERE, cLogger.getLevel());
+    assertEquals(0, cLogger.getHandlers().length);
+    assertTrue(aLogger.getUseParentHandlers());
+
+    List<String> logLines = retrieveLogLines();
+    assertThat(logLines, hasItem(containsString("Ignoring the direct logging level")));
   }
 
   @Test