get rid of the stuck status

confluentinc · Mar 20, 2020 · 5cb1af0 · 5cb1af0
1 parent b69ad12
commit 5cb1af0
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 13 deletions.
diff --git a/ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandRunner.java b/ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandRunner.java
@@ -70,7 +70,6 @@ public class CommandRunner implements Closeable {
 
   public enum CommandRunnerStatus {
     RUNNING,
-    STUCK,
     ERROR
   }
 
@@ -274,7 +273,7 @@ CommandRunnerStatus checkCommandRunnerStatus() {
 
     return Duration.between(currentCommand.right, clock.instant()).toMillis()
         < commandRunnerHealthTimeout.toMillis()
-        ? CommandRunnerStatus.RUNNING : CommandRunnerStatus.STUCK;
+        ? CommandRunnerStatus.RUNNING : CommandRunnerStatus.ERROR;
   }
 
   private class Runner implements Runnable {

diff --git a/...rc/test/java/io/confluent/ksql/rest/server/computation/CommandRunnerStatusMetricTest.java b/...rc/test/java/io/confluent/ksql/rest/server/computation/CommandRunnerStatusMetricTest.java
@@ -83,15 +83,6 @@ public void shouldInitiallyBeRunningState() {
     assertThat(currentGaugeValue(), is(CommandRunner.CommandRunnerStatus.RUNNING.name()));
   }
 
-  @Test
-  public void shouldUpdateToStuckState() {
-    // When:
-    when(commandRunner.checkCommandRunnerStatus()).thenReturn(CommandRunner.CommandRunnerStatus.STUCK);
-
-    // Then:
-    assertThat(currentGaugeValue(), is(CommandRunner.CommandRunnerStatus.STUCK.name()));
-  }
-
   @Test
   public void shouldUpdateErrorState() {
     // When:

diff --git a/...b-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/CommandRunnerTest.java b/...b-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/CommandRunnerTest.java
@@ -228,7 +228,7 @@ public void shouldEarlyOutIfNewCommandsContainsTerminate() {
   }
 
   @Test
-  public void shouldTransitionFromRunningToStuck() throws InterruptedException {
+  public void shouldTransitionFromRunningToErrorWhenStuckOnCommand() throws InterruptedException {
     // Given:
     givenQueuedCommands(queuedCommand1);
 
@@ -260,13 +260,27 @@ public void shouldTransitionFromRunningToStuck() throws InterruptedException {
     commandRunnerThread.start();
     commandSetLatch.await();
     assertThat(commandRunner.checkCommandRunnerStatus(), is(CommandRunner.CommandRunnerStatus.RUNNING));
-    assertThat(commandRunner.checkCommandRunnerStatus(), is(CommandRunner.CommandRunnerStatus.STUCK));
+    assertThat(commandRunner.checkCommandRunnerStatus(), is(CommandRunner.CommandRunnerStatus.ERROR));
     handleStatementLatch.countDown();
     commandRunnerThread.join();
     assertThat(commandRunner.checkCommandRunnerStatus(), is(CommandRunner.CommandRunnerStatus.RUNNING));
     assertThat(expectedException.get(), equalTo(null));
   }
 
+  @Test
+  public void shouldTransitionFromRunningToErrorWhenNotPollingCommandTopic() {
+    // Given:
+    givenQueuedCommands();
+
+    final Instant current = Instant.now();
+    when(clock.instant()).thenReturn(current)
+        .thenReturn(current.plusMillis(15100));
+
+    // Then:
+    commandRunner.fetchAndRunCommands();
+    assertThat(commandRunner.checkCommandRunnerStatus(), is(CommandRunner.CommandRunnerStatus.ERROR));
+  }
+
   @Test
   public void shouldEarlyOutOnShutdown() {
     // Given: