feat(cli): add the feature to turn of WRAP for CLI output

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

@@ -231,7 +231,7 @@ void handleLine(final String line) {
       return;
     }
 
-    handleStatements(line);
+    handleStatements(trimmedLine);
   }
 
   /**

ksql-cli/src/main/java/io/confluent/ksql/cli/console/CliConfig.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (the "License"; you may not use
+ * this file except in compliance with the License. You may obtain a copy of the
+ * License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.cli.console;
+
+import io.confluent.ksql.configdef.ConfigValidators;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.common.config.ConfigException;
+
+public class CliConfig extends AbstractConfig {
+
+  public static final String WRAP_CONFIG = "WRAP";
+
+  private static final ConfigDef CONFIG_DEF = new ConfigDef()
+      .define(
+          WRAP_CONFIG,
+          Type.STRING,
+          OnOff.ON.name(),
+          ConfigValidators.enumValues(OnOff.class),
+          Importance.MEDIUM,
+          "A value of 'OFF' will clip lines to ensure that query results do not exceed the "
+              + "terminal width (i.e. each row will appear on a single line)."
+      );
+
+  public CliConfig(final Map<?, ?> originals) {
+    super(CONFIG_DEF, originals);
+  }
+
+  public CliConfig with(final String property, final Object value) {
+    if (!CONFIG_DEF.names().contains(property.toUpperCase())) {
+      throw new ConfigException(
+          "Undefined property: " + property + ". Valid properties are: " + CONFIG_DEF.names());
+    }
+
+    final Map<String, Object> originals = new HashMap<>(originals());
+    originals.put(property.toUpperCase(), value);
+    return new CliConfig(originals);
+  }
+
+  @SuppressWarnings("unused") // used in validation
+  public enum OnOff {
+    ON, OFF
+  }
+
+}

ksql-cli/src/main/java/io/confluent/ksql/cli/console/Console.java

@@ -19,6 +19,7 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import io.confluent.ksql.GenericRow;
 import io.confluent.ksql.cli.console.KsqlTerminal.HistoryEntry;
@@ -105,6 +106,7 @@
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.ObjectUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
 import org.eclipse.jetty.io.RuntimeIOException;
 import org.jline.terminal.Terminal.Signal;
@@ -190,7 +192,7 @@ private static <T extends KsqlEntity> Handler1<KsqlEntity, Console> tablePrinter
   private final RowCaptor rowCaptor;
   private OutputFormat outputFormat;
   private Optional<File> spoolFile = Optional.empty();
-
+  private CliConfig config;
 
   public interface RowCaptor {
     void addRow(GenericRow row);
@@ -230,6 +232,7 @@ public Console(
     this.rowCaptor = Objects.requireNonNull(rowCaptor, "rowCaptor");
     this.cliSpecificCommands = Maps.newLinkedHashMap();
     this.objectMapper = JsonMapper.INSTANCE.mapper;
+    this.config = new CliConfig(ImmutableMap.of());
   }
 
   public PrintWriter writer() {
@@ -273,6 +276,14 @@ public void handle(final Signal signal, final SignalHandler signalHandler) {
     terminal.handle(signal, signalHandler);
   }
 
+  public void setCliProperty(final String name, final Object value) {
+    try {
+      config = config.with(name, value);
+    } catch (final ConfigException e) {
+      terminal.writer().println(e.getMessage());
+    }
+  }
+
   @Override
   public void close() {
     terminal.close();
@@ -407,14 +418,11 @@ private Optional<CliCmdExecutor> getCliCommand(final String line) {
       return Optional.empty();
     }
 
-    final String reconstructed = parts.stream()
-        .collect(Collectors.joining(" "));
-
-    final String asLowerCase = reconstructed.toLowerCase();
+    final String command = String.join(" ", parts);
 
-    return cliSpecificCommands.entrySet().stream()
-        .filter(e -> asLowerCase.startsWith(e.getKey()))
-        .map(e -> CliCmdExecutor.of(e.getValue(), parts))
+    return cliSpecificCommands.values().stream()
+        .filter(cliSpecificCommand -> cliSpecificCommand.matches(command))
+        .map(cliSpecificCommand -> CliCmdExecutor.of(cliSpecificCommand, parts))
         .findFirst();
   }
 
@@ -423,7 +431,7 @@ private void printAsTable(
       final List<FieldInfo> fields
   ) {
     rowCaptor.addRow(row);
-    writer().println(TabularRow.createRow(getWidth(), fields, row));
+    writer().println(TabularRow.createRow(getWidth(), fields, row, config));
     flush();
   }
 

ksql-cli/src/main/java/io/confluent/ksql/cli/console/cmd/CliCommandRegisterUtil.java

@@ -68,5 +68,9 @@ public static void registerDefaultCommands(
 
     console.registerCliSpecificCommand(
         Spool.create(console::setSpool, console::unsetSpool));
+
+    console.registerCliSpecificCommand(
+        SetCliProperty.create(console::setCliProperty)
+    );
   }
 }

ksql-cli/src/main/java/io/confluent/ksql/cli/console/cmd/CliSpecificCommand.java

@@ -20,6 +20,14 @@
 
 public interface CliSpecificCommand {
 
+  /**
+   * @param command the full command
+   * @return whether or not {@code command} is an instance of {@code this}
+   */
+  default boolean matches(final String command) {
+    return command.toLowerCase().startsWith(getName().toLowerCase());
+  }
+
   /**
    * Get the name of the command.
    *

ksql-cli/src/main/java/io/confluent/ksql/cli/console/cmd/SetCliProperty.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (the "License"; you may not use
+ * this file except in compliance with the License. You may obtain a copy of the
+ * License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.cli.console.cmd;
+
+import java.io.PrintWriter;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.BiConsumer;
+
+final class SetCliProperty implements CliSpecificCommand {
+
+  private static final String HELP = "set cli <property> <value>:" + System.lineSeparator()
+      + "\tSets a CLI local property. NOTE that this differs from setting a KSQL "
+      + "property with \"SET 'property'='value'\" in that it does not affect the server.";
+
+  private final BiConsumer<String, String> setProperty;
+
+  public static SetCliProperty create(final BiConsumer<String, String> setProperty) {
+    return new SetCliProperty(setProperty);
+  }
+
+  private SetCliProperty(final BiConsumer<String, String> setProperty) {
+    this.setProperty = Objects.requireNonNull(setProperty, "setProperty");
+  }
+
+  @Override
+  public String getName() {
+    return "set cli";
+  }
+
+  @Override
+  public String getHelpMessage() {
+    return HELP;
+  }
+
+  @Override
+  public void execute(final List<String> args, final PrintWriter terminal) {
+    CliCmdUtil.ensureArgCountBounds(args, 2, 2, getHelpMessage());
+    setProperty.accept(args.get(0), args.get(1));
+  }
+}

ksql-cli/src/main/java/io/confluent/ksql/util/TabularRow.java

@@ -19,6 +19,8 @@
 import com.google.common.base.Splitter;
 import com.google.common.base.Strings;
 import io.confluent.ksql.GenericRow;
+import io.confluent.ksql.cli.console.CliConfig;
+import io.confluent.ksql.cli.console.CliConfig.OnOff;
 import io.confluent.ksql.rest.entity.FieldInfo;
 import java.util.ArrayList;
 import java.util.List;
@@ -27,42 +29,49 @@
 
 public class TabularRow {
 
+  private static final String CLIPPED = "...";
   private static final int MIN_CELL_WIDTH = 5;
 
   private final int width;
   private final List<String> value;
   private final List<String> header;
   private final boolean isHeader;
+  private final boolean shouldWrap;
 
   public static TabularRow createHeader(final int width, final List<FieldInfo> header) {
     return new TabularRow(
         width,
         header.stream().map(FieldInfo::getName).collect(Collectors.toList()),
-        null);
+        null,
+        true);
   }
 
   public static TabularRow createRow(
       final int width,
       final List<FieldInfo> header,
-      final GenericRow value
+      final GenericRow value,
+      final CliConfig config
   ) {
     return new TabularRow(
         width,
         header.stream().map(FieldInfo::getName).collect(Collectors.toList()),
-        value.getColumns().stream().map(Objects::toString).collect(Collectors.toList())
+        value.getColumns().stream().map(Objects::toString).collect(Collectors.toList()),
+        config.getString(CliConfig.WRAP_CONFIG).equalsIgnoreCase(OnOff.ON.toString())
     );
   }
 
   @VisibleForTesting
   TabularRow(
       final int width,
       final List<String> header,
-      final List<String> value
+      final List<String> value,
+      final boolean shouldWrap
   ) {
     this.header = Objects.requireNonNull(header, "header");
     this.width = width;
     this.value = value;
     this.isHeader = value == null;
+    this.shouldWrap = shouldWrap;
   }
 
   @Override
@@ -92,7 +101,7 @@ public String toString() {
         .map(s -> addUntil(s, createCell("", cellWidth), maxSplit))
         .collect(Collectors.toList());
 
-    formatRow(builder, buffered, maxSplit);
+    formatRow(builder, buffered, shouldWrap ? maxSplit : 1);
 
     if (isHeader) {
       builder.append('\n');
@@ -111,14 +120,31 @@ private static void formatRow(
     for (int row = 0; row < numRows; row++) {
       builder.append('|');
       for (int col = 0; col < columns.size(); col++) {
-        builder.append(columns.get(col).get(row));
+        final String colValue = columns.get(col).get(row);
+        if (shouldClip(columns.get(col), numRows)) {
+          builder.append(colValue, 0, colValue.length() - CLIPPED.length())
+              .append(CLIPPED)
+              .append('|');
+        } else {
+          builder.append(colValue)
+              .append('|');
+        }
       }
       if (row != numRows - 1) {
         builder.append('\n');
       }
     }
   }
 
+  private static boolean shouldClip(final List<String> parts, final int rowsToPrint) {
+    // clip if there are more than one line and any of the remaining lines are non-empty
+    return parts.size() > rowsToPrint
+        && parts.subList(rowsToPrint, parts.size())
+        .stream()
+        .map(String::trim)
+        .noneMatch(String::isEmpty);
+  }
+
   @SuppressWarnings("UnstableApiUsage")
   private static List<String> splitToFixed(final String value, final int width) {
     return Splitter.fixedLength(width)
@@ -141,7 +167,7 @@ private static void separatingLine(
   }
 
   private static String createCell(final String value, final int width) {
-    final String format = "%-" + width + "s|";
+    final String format = "%-" + width + "s";
     return String.format(format, value);
   }
 

ksql-cli/src/test/java/io/confluent/ksql/cli/CliTest.java

@@ -458,6 +458,8 @@ public void testPropertySetUnset() {
     assertRunCommand("unset 'auto.offset.reset'", is(EMPTY_RESULT));
   }
 
+
+
   @Test
   public void shouldPrintCorrectSchemaForDescribeStream() {
     assertRunCommand(

ksql-cli/src/test/java/io/confluent/ksql/cli/console/ConsoleTest.java

@@ -135,6 +135,8 @@ public ConsoleTest(final OutputFormat outputFormat) {
     this.console = new Console(outputFormat, terminal, new NoOpRowCaptor());
 
     when(cliCommand.getName()).thenReturn(CLI_CMD_NAME);
+    when(cliCommand.matches(any()))
+        .thenAnswer(i -> ((String) i.getArgument(0)).toLowerCase().startsWith(CLI_CMD_NAME.toLowerCase()));
     console.registerCliSpecificCommand(cliCommand);
   }
 
@@ -1380,6 +1382,26 @@ public void shouldSwallowCliCommandLinesEvenWithWhiteSpace() {
     assertThat(result, is("not a CLI command;"));
   }
 
+  @Test
+  public void shouldThrowOnInvalidCliProperty() {
+    // When:
+    console.setCliProperty("FOO", "BAR");
+
+    // Then:
+    assertThat(terminal.getOutputString(),
+        containsString("Undefined property: FOO. Valid properties are"));
+  }
+
+  @Test
+  public void shouldThrowOnInvalidCliPropertyValue() {
+    // When:
+    console.setCliProperty(CliConfig.WRAP_CONFIG, "BURRITO");
+
+    // Then:
+    assertThat(terminal.getOutputString(),
+        containsString("Invalid value BURRITO for configuration WRAP: String must be one of: ON, OFF, null"));
+  }
+
   private static List<FieldInfo> buildTestSchema(final SqlType... fieldTypes) {
     final Builder schemaBuilder = LogicalSchema.builder();