update error messages

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandRunner.java

@@ -98,7 +98,8 @@ public enum CommandRunnerDegradedReason {
     NONE(errors -> ""),
     CORRUPTED(Errors::commandRunnerDegradedBackupCorruptedErrorMessage),
     INCOMPATIBLE_COMMAND(Errors::commandRunnerDegradedIncompatibleCommandsErrorMessage),
-    COMMAND_TOPIC_DELETED(Errors::commandRunnerDegradedCommandTopicDeletedErrorMessage);
+    COMMAND_TOPIC_DELETED_OR_MODIFIED(
+        Errors::commandRunnerDegradedCommandTopicDeletedModifiedErrorMessage);
 
     private final Function<Errors, String> msgFactory;
 
@@ -445,12 +446,12 @@ public void run() {
             );
             closeEarly();
           } else if (commandTopicDeleted) {
-              LOG.warn("CommandRunner entering degraded state due to command topic deletion.");
-              state = new Status(
-                  CommandRunnerStatus.DEGRADED,
-                  CommandRunnerDegradedReason.COMMAND_TOPIC_DELETED
-              );
-              closeEarly();
+            LOG.warn("CommandRunner entering degraded state due to command topic deletion.");
+            state = new Status(
+                CommandRunnerStatus.DEGRADED,
+                CommandRunnerDegradedReason.COMMAND_TOPIC_DELETED_OR_MODIFIED
+            );
+            closeEarly();
           } else {
             LOG.trace("Polling for new writes to command topic");
             fetchAndRunCommands();
@@ -464,7 +465,7 @@ public void run() {
         LOG.warn("The command topic offset was reset. CommandRunner thread exiting.");
         state = new Status(
             CommandRunnerStatus.DEGRADED,
-            CommandRunnerDegradedReason.COMMAND_TOPIC_DELETED
+            CommandRunnerDegradedReason.COMMAND_TOPIC_DELETED_OR_MODIFIED
         );
         closeEarly();
       } finally {

ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/CommandRunnerTest.java

@@ -135,7 +135,7 @@ public void setup() {
     when(compactor.apply(any())).thenAnswer(inv -> inv.getArgument(0));
     when(errorHandler.commandRunnerDegradedIncompatibleCommandsErrorMessage()).thenReturn(INCOMPATIBLE_COMMANDS_ERROR_MESSAGE);
     when(errorHandler.commandRunnerDegradedBackupCorruptedErrorMessage()).thenReturn(BACKUP_CORRUPTED_ERROR_MESSAGE);
-    when(errorHandler.commandRunnerDegradedCommandTopicDeletedErrorMessage()).thenReturn(MISSING_COMMAND_TOPIC_ERROR_MESSAGE);
+    when(errorHandler.commandRunnerDegradedCommandTopicDeletedModifiedErrorMessage()).thenReturn(MISSING_COMMAND_TOPIC_ERROR_MESSAGE);
 
     givenQueuedCommands(queuedCommand1, queuedCommand2, queuedCommand3);
 
@@ -355,7 +355,9 @@ public void shouldEnterDegradedStateIfCommandTopicMissing() {
 
     assertThat(commandRunner.checkCommandRunnerStatus(), is(CommandRunner.CommandRunnerStatus.DEGRADED));
     assertThat(commandRunner.getCommandRunnerDegradedWarning(), is(MISSING_COMMAND_TOPIC_ERROR_MESSAGE));
-    assertThat(commandRunner.getCommandRunnerDegradedReason(), is(CommandRunner.CommandRunnerDegradedReason.COMMAND_TOPIC_DELETED));
+    assertThat(
+        commandRunner.getCommandRunnerDegradedReason(),
+        is(CommandRunner.CommandRunnerDegradedReason.COMMAND_TOPIC_DELETED_OR_MODIFIED));
   }
 
   @Test
@@ -549,6 +551,7 @@ public void shouldCloseEarlyWhenSerializationExceptionInFetch() throws Exception
     inOrder.verify(commandStore).close();
   }
 
+  @Test
   public void shouldCloseEarlyWhenOffsetOutOfRangeException() throws Exception {
     // Given:
     when(commandStore.getNewCommands(any()))

ksqldb-rest-model/src/main/java/io/confluent/ksql/rest/DefaultErrorMessages.java

@@ -39,8 +39,8 @@ public class DefaultErrorMessages implements ErrorMessages {
           + "The server must be restarted after performing either operation in order to resume "
           + "normal functionality";
 
-  public static final String COMMAND_RUNNER_DEGRADED_COMMAND_TOPIC_DELETED =
-      "The server is in a degraded state due to deletion of the command topic. "
+  public static final String COMMAND_RUNNER_DEGRADED_COMMAND_TOPIC_DELETED_MODIFIED =
+      "The server has detected that the command topic has been modified or deleted."
           + "DDL statements will not be processed."
           + System.lineSeparator()
           + "Restart the server to restore server functionality.";
@@ -81,7 +81,7 @@ public String commandRunnerDegradedBackupCorruptedErrorMessage() {
   }
 
   @Override
-  public String commandRunnerDegradedCommandTopicDeletedErrorMessage() {
-    return COMMAND_RUNNER_DEGRADED_COMMAND_TOPIC_DELETED;
+  public String commandRunnerDegradedCommandTopicDeletedModifiedErrorMessage() {
+    return COMMAND_RUNNER_DEGRADED_COMMAND_TOPIC_DELETED_MODIFIED;
   }
 }

ksqldb-rest-model/src/main/java/io/confluent/ksql/rest/ErrorMessages.java

@@ -27,5 +27,5 @@ public interface ErrorMessages {
 
   String commandRunnerDegradedBackupCorruptedErrorMessage();
 
-  String commandRunnerDegradedCommandTopicDeletedErrorMessage();
+  String commandRunnerDegradedCommandTopicDeletedModifiedErrorMessage();
 }

ksqldb-rest-model/src/main/java/io/confluent/ksql/rest/Errors.java

@@ -234,8 +234,8 @@ public String commandRunnerDegradedBackupCorruptedErrorMessage() {
     return errorMessages.commandRunnerDegradedBackupCorruptedErrorMessage();
   }
 
-  public String commandRunnerDegradedCommandTopicDeletedErrorMessage() {
-    return errorMessages.commandRunnerDegradedCommandTopicDeletedErrorMessage();
+  public String commandRunnerDegradedCommandTopicDeletedModifiedErrorMessage() {
+    return errorMessages.commandRunnerDegradedCommandTopicDeletedModifiedErrorMessage();
   }
 
   public EndpointResponse generateResponse(