diff --git a/ksql-cli/src/main/java/io/confluent/ksql/cli/console/table/Table.java b/ksql-cli/src/main/java/io/confluent/ksql/cli/console/table/Table.java
index 26d5231780b..c1bb801f24b 100644
--- a/ksql-cli/src/main/java/io/confluent/ksql/cli/console/table/Table.java
+++ b/ksql-cli/src/main/java/io/confluent/ksql/cli/console/table/Table.java
@@ -130,7 +130,8 @@ public void print(final Console console) {
 
       console.writer().printf(rowFormatString, columnHeaders.toArray());
 
-      final String separator = StringUtils.repeat('-', separatorLength);
+      final String separator =
+          StringUtils.repeat('-', Math.min(console.getWidth(), separatorLength));
       console.writer().println(separator);
       for (final List<String> row : rowValues) {
         console.writer().printf(rowFormatString, row.toArray());
diff --git a/ksql-cli/src/main/java/io/confluent/ksql/cli/console/table/builder/ErrorEntityTableBuilder.java b/ksql-cli/src/main/java/io/confluent/ksql/cli/console/table/builder/ErrorEntityTableBuilder.java
index 5aca7eb1454..5242cc05955 100644
--- a/ksql-cli/src/main/java/io/confluent/ksql/cli/console/table/builder/ErrorEntityTableBuilder.java
+++ b/ksql-cli/src/main/java/io/confluent/ksql/cli/console/table/builder/ErrorEntityTableBuilder.java
@@ -15,16 +15,33 @@
 
 package io.confluent.ksql.cli.console.table.builder;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Splitter;
 import io.confluent.ksql.cli.console.table.Table;
+import io.confluent.ksql.json.JsonMapper;
 import io.confluent.ksql.rest.entity.ErrorEntity;
+import java.io.IOException;
 
 public class ErrorEntityTableBuilder implements TableBuilder<ErrorEntity> {
 
+  private static final ObjectMapper MAPPER = JsonMapper.INSTANCE.mapper;
+
   @Override
   public Table buildTable(final ErrorEntity entity) {
+    final String message = entity.getErrorMessage();
+
+    String formatted;
+    try {
+      formatted = MAPPER
+          .writerWithDefaultPrettyPrinter()
+          .writeValueAsString(MAPPER.readTree(message));
+    } catch (IOException e) {
+      formatted = String.join("\n", Splitter.fixedLength(60).splitToList(message));
+    }
+
     return new Table.Builder()
         .withColumnHeaders("Error")
-        .withRow(entity.getErrorMessage())
+        .withRow(formatted)
         .build();
   }
 }
diff --git a/ksql-cli/src/test/java/io/confluent/ksql/cli/console/ConsoleTest.java b/ksql-cli/src/test/java/io/confluent/ksql/cli/console/ConsoleTest.java
index 5b7b8aba784..7aac649b794 100644
--- a/ksql-cli/src/test/java/io/confluent/ksql/cli/console/ConsoleTest.java
+++ b/ksql-cli/src/test/java/io/confluent/ksql/cli/console/ConsoleTest.java
@@ -25,7 +25,10 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import io.confluent.ksql.FakeException;
 import io.confluent.ksql.GenericRow;
@@ -39,6 +42,7 @@
 import io.confluent.ksql.rest.entity.ConnectorDescription;
 import io.confluent.ksql.rest.entity.ConnectorList;
 import io.confluent.ksql.rest.entity.EntityQueryId;
+import io.confluent.ksql.rest.entity.ErrorEntity;
 import io.confluent.ksql.rest.entity.ExecutionPlan;
 import io.confluent.ksql.rest.entity.FieldInfo;
 import io.confluent.ksql.rest.entity.FunctionDescriptionList;
@@ -71,6 +75,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
@@ -1034,6 +1039,61 @@ public void shouldPrintWarnings() throws IOException {
     }
   }
 
+  @Test
+  public void shouldPrintErrorEntityLongNonJson() throws IOException {
+    // Given:
+    final KsqlEntity entity = new ErrorEntity(
+        "statementText",
+        Strings.repeat("Not a JSON value! ", 10));
+
+    // When:
+    console.printKsqlEntityList(ImmutableList.of(entity));
+
+    // Then:
+    final String output = terminal.getOutputString();
+    if (console.getOutputFormat() == OutputFormat.TABULAR) {
+      assertThat(
+          output,
+          is("\n"
+              + " Error                                                        \n"
+              + "--------------------------------------------------------------\n"
+              + " Not a JSON value! Not a JSON value! Not a JSON value! Not a \n"
+              + "JSON value! Not a JSON value! Not a JSON value! Not a JSON v\n"
+              + "alue! Not a JSON value! Not a JSON value! Not a JSON value!  \n"
+              + "--------------------------------------------------------------\n")
+      );
+    }
+  }
+
+  @Test
+  public void shouldPrintErrorEntityLongJson() throws IOException {
+    // Given:
+    final KsqlEntity entity = new ErrorEntity(
+        "statementText",
+        new ObjectMapper().writeValueAsString(ImmutableMap.of(
+            "foo", "bar",
+            "message", "a " + StringUtils.repeat("really ", 20) + " long message"
+        )));
+
+    // When:
+    console.printKsqlEntityList(ImmutableList.of(entity));
+
+    // Then:
+    final String output = terminal.getOutputString();
+    if (console.getOutputFormat() == OutputFormat.TABULAR) {
+      assertThat(
+          output,
+          containsString(""
+              + "----------------------------------------------------------------------------------------------------\n"
+              + " {\n"
+              + "  \"foo\" : \"bar\",\n"
+              + "  \"message\" : \"a really really really really really really really really really really really really really really really really really really really really  long message\"\n"
+              + "} \n"
+              + "----------------------------------------------------------------------------------------------------")
+      );
+    }
+  }
+
   @Test
   public void shouldPrintFunctionDescription() throws IOException {
     final KsqlEntityList entityList = new KsqlEntityList(ImmutableList.of(