Repository: googlecloudplatform/dataflowtemplates
Issue: [Bug]: <Please Replace with a Summary of the Bug> 

### Related Template(s)

MQTT to Pub/Sub template

### What happened?

Providing both username and password throws an IllegalArgumentException.

### Beam Version

Newer than 2.46.0

### Relevant log output

```shell
While username is provided, password is required for authentication
```


Changed files:
v2/mqtt-to-pubsub/src/main/java/com/google/cloud/teleport/v2/templates/MqttToPubsub.java
v2/mqtt-to-pubsub/src/test/java/com/google/cloud/teleport/v2/templates/MqttToPubsubTest.java

diff --git a/v2/mqtt-to-pubsub/src/main/java/com/google/cloud/teleport/v2/templates/MqttToPubsub.java b/v2/mqtt-to-pubsub/src/main/java/com/google/cloud/teleport/v2/templates/MqttToPubsub.java
index 37f98fbc..9cbf6b21 100644
--- a/v2/mqtt-to-pubsub/src/main/java/com/google/cloud/teleport/v2/templates/MqttToPubsub.java
+++ b/v2/mqtt-to-pubsub/src/main/java/com/google/cloud/teleport/v2/templates/MqttToPubsub.java
@@ -15,6 +15,9 @@
  */
 package com.google.cloud.teleport.v2.templates;
 
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Strings.isNullOrEmpty;
+
 import com.google.cloud.teleport.metadata.Template;
 import com.google.cloud.teleport.metadata.TemplateCategory;
 import com.google.cloud.teleport.metadata.TemplateParameter;
@@ -58,15 +61,25 @@ public class MqttToPubsub {
   }
 
   public static void validate(MqttToPubsubOptions options) {
-    if (options != null) {
-      if ((options.getUsername() != null
-              && (!options.getUsername().isEmpty() || !options.getUsername().isBlank()))
-          && (options.getPassword() != null
-              || options.getPassword().isBlank()
-              || options.getPassword().isEmpty())) {
-        throw new IllegalArgumentException(
-            "While username is provided, password is required for authentication");
-      }
+    checkArgument(
+        options != null,
+        String.format("%s is required to run this template", MqttToPubsubOptions.class.getName()));
+    validateCredentials(options);
+  }
+
+  private static void validateCredentials(MqttToPubsubOptions options) {
+
+    boolean isUsernameNullOrEmpty = isNullOrEmpty(options.getUsername());
+    boolean isPasswordNullOrEmpty = isNullOrEmpty(options.getPassword());
+
+    // Validation passes when both username and password are provided together or neither provided.
+    // Uses xor operator '^' to simplify this evaluation. The following conditional is true when
+    // either are true but not when both are true or both are false.
+    if (isUsernameNullOrEmpty ^ isPasswordNullOrEmpty) {
+      throw new IllegalArgumentException(
+          String.format(
+              "%s expects either both a username and password or neither",
+              MqttToPubsubOptions.class.getName()));
     }
   }
 
@@ -99,6 +112,7 @@ public class MqttToPubsub {
   }
 
   static class ByteToStringTransform extends DoFn<byte[], String> {
+
     @ProcessElement
     public void processElement(@Element byte[] word, OutputReceiver<String> out) {
       out.output(new String(word, StandardCharsets.UTF_8));
@@ -110,6 +124,7 @@ public class MqttToPubsub {
    * executor at the command-line.
    */
   public interface MqttToPubsubOptions extends PipelineOptions {
+
     @TemplateParameter.Text(
         order = 1,
         optional = true,
@@ -138,7 +153,8 @@ public class MqttToPubsub {
         order = 3,
         description = "Output Pub/Sub topic",
         helpText =
-            "The name of the topic to which data should published, in the format of 'projects/your-project-id/topics/your-topic-name'",
+            "The name of the topic to which data should published, in the format of"
+                + " 'projects/your-project-id/topics/your-topic-name'",
         example = "projects/your-project-id/topics/your-topic-name")
     @Validation.Required
     String getOutputTopic();
diff --git a/v2/mqtt-to-pubsub/src/test/java/com/google/cloud/teleport/v2/templates/MqttToPubsubTest.java b/v2/mqtt-to-pubsub/src/test/java/com/google/cloud/teleport/v2/templates/MqttToPubsubTest.java
index 3f68e5a2..73b9b8f7 100644
--- a/v2/mqtt-to-pubsub/src/test/java/com/google/cloud/teleport/v2/templates/MqttToPubsubTest.java
+++ b/v2/mqtt-to-pubsub/src/test/java/com/google/cloud/teleport/v2/templates/MqttToPubsubTest.java
@@ -15,6 +15,7 @@
  */
 package com.google.cloud.teleport.v2.templates;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThrows;
 
 import java.nio.charset.StandardCharsets;
@@ -51,11 +52,94 @@ public class MqttToPubsubTest {
   }
 
   @Test
-  public void testValidation() {
+  public void testValidation_emptyPassword_throwsIllegalArgumentException() {
     MqttToPubsub.MqttToPubsubOptions options =
         PipelineOptionsFactory.create().as(MqttToPubsub.MqttToPubsubOptions.class);
     options.setUsername("test");
     options.setPassword("");
-    assertThrows(IllegalArgumentException.class, () -> MqttToPubsub.validate(options));
+
+    IllegalArgumentException e =
+        assertThrows(IllegalArgumentException.class, () -> MqttToPubsub.validate(options));
+
+    assertEquals(
+        "com.google.cloud.teleport.v2.templates.MqttToPubsub$MqttToPubsubOptions expects either both a username and password or neither",
+        e.getMessage());
+  }
+
+  @Test
+  public void testValidation_nullPassword_throwsIllegalArgumentException() {
+    MqttToPubsub.MqttToPubsubOptions options =
+        PipelineOptionsFactory.create().as(MqttToPubsub.MqttToPubsubOptions.class);
+    options.setUsername("test");
+    options.setPassword(null);
+
+    IllegalArgumentException e =
+        assertThrows(IllegalArgumentException.class, () -> MqttToPubsub.validate(options));
+
+    assertEquals(
+        "com.google.cloud.teleport.v2.templates.MqttToPubsub$MqttToPubsubOptions expects either both a username and password or neither",
+        e.getMessage());
+  }
+
+  @Test
+  public void testValidation_emptyUsername_throwsIllegalArgumentException() {
+    MqttToPubsub.MqttToPubsubOptions options =
+        PipelineOptionsFactory.create().as(MqttToPubsub.MqttToPubsubOptions.class);
+    options.setUsername("");
+    options.setPassword("password");
+
+    IllegalArgumentException e =
+        assertThrows(IllegalArgumentException.class, () -> MqttToPubsub.validate(options));
+
+    assertEquals(
+        "com.google.cloud.teleport.v2.templates.MqttToPubsub$MqttToPubsubOptions expects either both a username and password or neither",
+        e.getMessage());
+  }
+
+  @Test
+  public void testValidation_nullUsername_throwsIllegalArgumentException() {
+    MqttToPubsub.MqttToPubsubOptions options =
+        PipelineOptionsFactory.create().as(MqttToPubsub.MqttToPubsubOptions.class);
+    options.setUsername(null);
+    options.setPassword("password");
+
+    IllegalArgumentException e =
+        assertThrows(IllegalArgumentException.class, () -> MqttToPubsub.validate(options));
+
+    assertEquals(
+        "com.google.cloud.teleport.v2.templates.MqttToPubsub$MqttToPubsubOptions expects either both a username and password or neither",
+        e.getMessage());
+  }
+
+  @Test
+  public void testValidation_nonEmptyUsernamePassword_passes() {
+    MqttToPubsub.MqttToPubsubOptions options =
+        PipelineOptionsFactory.create().as(MqttToPubsub.MqttToPubsubOptions.class);
+    options.setUsername("username");
+    options.setPassword("password");
+
+    // validates with no errors
+    MqttToPubsub.validate(options);
+  }
+
+  @Test
+  public void testValidation_emptyUsernamePassword_passes() {
+    MqttToPubsub.MqttToPubsubOptions options =
+        PipelineOptionsFactory.create().as(MqttToPubsub.MqttToPubsubOptions.class);
+    options.setUsername("");
+    options.setPassword("");
+
+    // validates with no errors
+    MqttToPubsub.validate(options);
+  }
+
+  @Test
+  public void run_nullMqttToPubsubOptions_throwsIllegalStateException() {
+    IllegalArgumentException e =
+        assertThrows(IllegalArgumentException.class, () -> MqttToPubsub.run(null));
+
+    assertEquals(
+        "com.google.cloud.teleport.v2.templates.MqttToPubsub$MqttToPubsubOptions is required to run this template",
+        e.getMessage());
   }
 }