refactor(stream-platform): remove unnecessary single method call

(cherry picked from commit a436243)
camunda · Feb 26, 2024 · 7c8039b · 7c8039b
1 parent 01cef5b
commit 7c8039b
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 11 deletions.
diff --git a/stream-platform/src/main/java/io/camunda/zeebe/stream/impl/ProcessingStateMachine.java b/stream-platform/src/main/java/io/camunda/zeebe/stream/impl/ProcessingStateMachine.java
@@ -60,7 +60,7 @@
  *
  * +------------------+            +--------------------+
  * |                  |            |                    |      exception
- * | readNextRecord() |----------->|  processCommand()  |------------------+
+ * | tryToReadNextRecord() |----------->|  processCommand()  |------------------+
  * |                  |            |                    |                  v
  * +------------------+            +--------------------+            +---------------+
  *           ^                             |                         |               |------+
@@ -201,14 +201,10 @@ public ProcessingStateMachine(
   private void skipRecord() {
     notifySkippedListener(currentRecord);
     markProcessingCompleted();
-    actor.submit(this::readNextRecord);
+    actor.submit(this::tryToReadNextRecord);
     metrics.eventSkipped();
   }
 
-  void readNextRecord() {
-    tryToReadNextRecord();
-  }
-
   void markProcessingCompleted() {
     inProcessing = false;
     if (onErrorRetries > 0) {
@@ -217,7 +213,7 @@ void markProcessingCompleted() {
     }
   }
 
-  private void tryToReadNextRecord() {
+  void tryToReadNextRecord() {
     final var hasNext = logStreamReader.hasNext();
 
     if (currentRecord != null) {
@@ -700,7 +696,7 @@ private void executeSideEffects() {
 
           // continue with next record
           markProcessingCompleted();
-          actor.submit(this::readNextRecord);
+          actor.submit(this::tryToReadNextRecord);
         });
   }
 
@@ -753,7 +749,7 @@ public void startProcessing(final LastProcessingPositions lastProcessingPosition
       lastWrittenPosition = lastProcessingPositions.getLastWrittenPosition();
     }
 
-    actor.submit(this::readNextRecord);
+    actor.submit(this::tryToReadNextRecord);
   }
 
   private record BatchProcessingStepResult(

diff --git a/stream-platform/src/main/java/io/camunda/zeebe/stream/impl/StreamProcessor.java b/stream-platform/src/main/java/io/camunda/zeebe/stream/impl/StreamProcessor.java
@@ -563,7 +563,7 @@ public void resumeProcessing() {
               lifecycleAwareListeners.forEach(StreamProcessorLifecycleAware::onResumed);
               streamProcessorContext.streamProcessorPhase(Phase.PROCESSING);
               if (processingStateMachine != null) {
-                actor.submit(processingStateMachine::readNextRecord);
+                actor.submit(processingStateMachine::tryToReadNextRecord);
               }
               LOG.debug("Resumed processing for partition {}", partitionId);
             }
@@ -573,7 +573,7 @@ public void resumeProcessing() {
 
   @Override
   public void onRecordAvailable() {
-    actor.run(processingStateMachine::readNextRecord);
+    actor.run(processingStateMachine::tryToReadNextRecord);
   }
 
   private static final class AsyncProcessingScheduleServiceActor extends Actor {