fix: fix NPE in CLI if not username supplied

Fixes NPE when running the CLI with no arguments. Was throwing NPE on line:

```java
if (!options.getUserName().isEmpty() && !options.isPasswordSet()) {
```

in Ksql.java, as `options.getUserName()` was returning `null`.

Refactored so that username and password are never null, only empty.

Also refactored so that server will default to `http://localhost:8088`, even if other params are supplied.  Previously, it only defaulted to this if no params supplied.

ksql-cli/src/main/java/io/confluent/ksql/Ksql.java

@@ -25,7 +25,6 @@
 import io.confluent.ksql.util.ErrorMessageUtil;
 import io.confluent.ksql.version.metrics.KsqlVersionCheckerAgent;
 import io.confluent.ksql.version.metrics.collector.KsqlModuleType;
-
 import java.io.Console;
 import java.io.File;
 import java.io.IOException;
@@ -61,16 +60,13 @@ public final class Ksql {
   }
 
   public static void main(final String[] args) throws IOException {
-    final Options options = args.length == 0
-        ? Options.parse("http://localhost:8088")
-        : Options.parse(args);
-
+    final Options options = Options.parse(args);
     if (options == null) {
       System.exit(-1);
     }
 
     // ask for password if not set through command parameters
-    if (!options.getUserName().isEmpty() && !options.isPasswordSet()) {
+    if (options.requiresPassword()) {
       options.setPassword(readPassword());
     }
 
@@ -90,8 +86,15 @@ private static String readPassword() {
       System.err.println("Could not get console for enter password; use -p option instead.");
       System.exit(-1);
     }
-
-    return new String(console.readPassword("Enter password: "));
+
+    String password = "";
+    while (password.isEmpty()) {
+      password = new String(console.readPassword("Enter password: "));
+      if (password.isEmpty()) {
+        console.writer().println("Error: password can not be empty");
+      }
+    }
+    return password;
   }
 
   void run() {

ksql-cli/src/main/java/io/confluent/ksql/cli/Options.java

@@ -21,7 +21,6 @@
 import com.github.rvesse.airline.annotations.Command;
 import com.github.rvesse.airline.annotations.Option;
 import com.github.rvesse.airline.annotations.restrictions.Once;
-import com.github.rvesse.airline.annotations.restrictions.Required;
 import com.github.rvesse.airline.annotations.restrictions.ranges.LongRange;
 import com.github.rvesse.airline.help.Help;
 import com.github.rvesse.airline.parser.errors.ParseException;
@@ -47,13 +46,12 @@ public class Options {
   @Inject
   public HelpOption<?> help;
 
-  @SuppressWarnings("unused") // Accessed via reflection
+  @SuppressWarnings({"unused", "FieldMayBeFinal", "FieldCanBeLocal"}) // Accessed via reflection
   @Once
-  @Required
   @Arguments(
       title = "server",
       description = "The address of the Ksql server to connect to (ex: http://confluent.io:9098)")
-  private String server;
+  private String server = "http://localhost:8088";
 
   private static final String CONFIGURATION_FILE_OPTION_NAME = "--config-file";
 
@@ -65,7 +63,7 @@ public class Options {
   private String configFile;
 
 
-  @SuppressWarnings("unused") // Accessed via reflection
+  @SuppressWarnings({"unused", "FieldMayBeFinal"}) // Accessed via reflection
   @Option(
       name = {USERNAME_OPTION, USERNAME_SHORT_OPTION},
       description =
@@ -75,7 +73,7 @@ public class Options {
               + "/"
               + PASSWORD_OPTION
               + " flag")
-  private String userName;
+  private String userName = "";
 
   @SuppressWarnings("unused") // Accessed via reflection
   @Option(
@@ -87,7 +85,7 @@ public class Options {
               + "/"
               + USERNAME_OPTION
               + " flag")
-  private String password;
+  private String password = "";
 
   @SuppressWarnings("unused") // Accessed via reflection
   @Option(
@@ -158,20 +156,24 @@ public OutputFormat getOutputFormat() {
     return OutputFormat.valueOf(outputFormat);
   }
 
-  public String getUserName() {
-    return userName;
+  public boolean requiresPassword() {
+    if (userName.isEmpty()) {
+      return false;
+    }
+
+    return password.trim().isEmpty();
   }
 
   public void setPassword(final String password) {
-    this.password = password;
-  }
+    if (password.isEmpty()) {
+      throw new IllegalArgumentException("Password must not be empty");
+    }
 
-  public boolean isPasswordSet() {
-    return (password != null && !password.trim().isEmpty());
+    this.password = password;
   }
 
   public Optional<BasicCredentials> getUserNameAndPassword() {
-    if ((userName == null && password != null) || (password == null && userName != null)) {
+    if (userName.isEmpty() != password.isEmpty()) {
       throw new ConfigException(
           "You must specify both a username and a password. If you don't want to use an "
               + "authenticated session, don't specify either of the "
@@ -181,7 +183,7 @@ public Optional<BasicCredentials> getUserNameAndPassword() {
               + " flags on the command line");
     }
 
-    if (userName == null) {
+    if (userName.isEmpty()) {
       return Optional.empty();
     }
 

ksql-cli/src/test/java/io/confluent/ksql/cli/commands/OptionsTest.java

@@ -15,55 +15,140 @@
 
 package io.confluent.ksql.cli.commands;
 
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
 
 import io.confluent.ksql.cli.Options;
-
+import io.confluent.ksql.rest.client.BasicCredentials;
+import java.util.Optional;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.common.config.ConfigException;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 
 public class OptionsTest {
 
-  @Test(expected = ConfigException.class)
-  public void shouldThrowConfigExceptionIfOnlyUsernameIsProvided() throws Exception {
-    final Options options = Options.parse("http://foobar", "-u", "joe");
+  @Rule
+  public final ExpectedException expectedException = ExpectedException.none();
+
+  @Test
+  public void shouldUseDefaultServerIfNoneSupplied() {
+    // When:
+    final Options options = parse();
+
+    // Then:
+    assertThat(options.getServer(), is("http://localhost:8088"));
+  }
+
+  @Test
+  public void shouldWorkWithUserSuppliedServer() {
+    // When:
+    final Options options = parse("custom server");
+
+    // Then:
+    assertThat(options.getServer(), is("custom server"));
+  }
+
+  @Test
+  public void shouldThrowConfigExceptionIfOnlyUsernameIsProvided() {
+    // Given:
+    final Options options = parse("-u", "joe");
+
+    // Expect:
+    expectedException.expect(ConfigException.class);
+    expectedException.expectMessage("You must specify both a username and a password");
+
+    // When:
     options.getUserNameAndPassword();
   }
 
-  @Test(expected = ConfigException.class)
-  public void shouldThrowConfigExceptionIfOnlyPasswordIsProvided() throws Exception {
-    final Options options = Options.parse("http://foobar", "-p", "joe");
+  @Test
+  public void shouldThrowConfigExceptionIfOnlyPasswordIsProvided() {
+    // Given:
+    final Options options = parse("http://foobar", "-p", "joe");
+
+    // Expect:
+    expectedException.expect(ConfigException.class);
+    expectedException.expectMessage("You must specify both a username and a password");
+
+    // When:
     options.getUserNameAndPassword();
   }
 
   @Test
-  public void shouldReturnUserPasswordPairWhenBothProvided() throws Exception {
-    final Options options = Options.parse("http://foobar", "-u", "joe", "-p", "joe");
-    assertTrue(options.getUserNameAndPassword().isPresent());
+  public void shouldReturnUserPasswordPairWhenBothProvided() {
+    // When:
+    final Options options = parse("http://foobar", "-u", "joe", "-p", "pp");
+
+    // Then:
+    assertThat(options.getUserNameAndPassword(),
+        is(Optional.of(BasicCredentials.of("joe", "pp"))));
+  }
+
+  @Test
+  public void shouldReturnEmptyOptionWhenUserAndPassNotPresent() {
+    // When:
+    final Options options = parse();
+
+    // Then:
+    assertThat(options.getUserNameAndPassword(), is(Optional.empty()));
   }
 
   @Test
-  public void shouldReturnEmptyOptionWhenUserAndPassNotPresent() throws Exception {
-    final Options options = Options.parse("http://foobar");
-    assertFalse(options.getUserNameAndPassword().isPresent());
+  public void shouldNotRequirePasswordIfUserNameNotSet() {
+    // When:
+    final Options options = parse();
+
+    // Then:
+    assertThat(options.requiresPassword(), is(false));
   }
 
   @Test
-  public void shouldReturnPasswordNotSetIfPasswordIsNull() throws Exception {
-    final Options options = Options.parse("http://foobar");
-    assertFalse(options.isPasswordSet());
+  public void shouldNotRequirePasswordIfUserNameAndPasswordSupplied() {
+    // When:
+    final Options options = parse("-u", "joe", "-p", "oo");
+
+    // Then:
+    assertThat(options.requiresPassword(), is(false));
   }
 
   @Test
-  public void shouldReturnPasswordNotSetIfPasswordIsEmpty() throws Exception {
-    final Options options = Options.parse("http://foobar", "-u", "joe", "-p", "");
-    assertFalse(options.isPasswordSet());
+  public void shouldRequirePasswordIfUserNameSuppliedButNotPassword() {
+    // When:
+    final Options options = parse("-u", "joe");
+
+    // Then:
+    assertThat(options.requiresPassword(), is(true));
   }
 
   @Test
-  public void shouldReturnPasswordSetIfPasswordIsNotEmpty() throws Exception {
-    final Options options = Options.parse("http://foobar", "-u", "joe", "-p", "joe");
-    assertTrue(options.isPasswordSet());
+  public void shouldNotRequirePasswordIfUserNameAndPasswordSuppliedButEmpty() {
+    // When:
+    final Options options = parse("-u", "joe", "-p", "");
+
+    // Then:
+    assertThat(options.requiresPassword(), is(true));
+  }
+
+  @Test
+  public void shouldNotTrimPasswords() {
+    // When:
+    final Options options = parse("-u", "joe", "-p", "  ");
+
+    // Then:
+    assertThat(options.getUserNameAndPassword().map(BasicCredentials::password),
+        is(Optional.of("  ")));
+  }
+
+  private static Options parse(final String... args) {
+    try {
+      final Options parsed = Options.parse(args);
+      assertThat(parsed, is(notNullValue()));
+      return parsed;
+    } catch (final Exception e) {
+      throw new AssertionError("Failed to parse options: " + StringUtils.join(args, ","), e);
+    }
   }
 }