[Backport stable/8.0] Testcontainer engine lifecycle

engine-agent/src/main/java/io/camunda/zeebe/process/test/engine/agent/EngineControlImpl.java

@@ -29,13 +29,9 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.TimeoutException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public final class EngineControlImpl extends EngineControlImplBase {
 
-  private static final Logger LOG = LoggerFactory.getLogger(EngineControlImpl.class);
-
   private ZeebeTestEngine engine;
   private RecordStreamSourceWrapper recordStreamSource;
 
@@ -49,6 +45,7 @@ public void startEngine(
       final StartEngineRequest request,
       final StreamObserver<StartEngineResponse> responseObserver) {
     engine.start();
+    recordStreamSource = new RecordStreamSourceWrapper(engine.getRecordStreamSource());
 
     final StartEngineResponse response = StartEngineResponse.newBuilder().build();
     responseObserver.onNext(response);
@@ -71,8 +68,6 @@ public void resetEngine(
       final StreamObserver<ResetEngineResponse> responseObserver) {
     engine.stop();
     engine = EngineFactory.create(AgentProperties.getGatewayPort());
-    recordStreamSource = new RecordStreamSourceWrapper(engine.getRecordStreamSource());
-    engine.start();
 
     final ResetEngineResponse response = ResetEngineResponse.newBuilder().build();
     responseObserver.onNext(response);

engine-agent/src/main/java/io/camunda/zeebe/process/test/engine/agent/ZeebeProcessTestEngine.java

@@ -12,9 +12,13 @@
 import io.grpc.Server;
 import io.grpc.ServerBuilder;
 import java.io.IOException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class ZeebeProcessTestEngine {
 
+  private static final Logger LOG = LoggerFactory.getLogger(ZeebeProcessTestEngine.class);
+
   public static void main(final String[] args) throws IOException {
     final ZeebeTestEngine engine = EngineFactory.create(AgentProperties.getGatewayPort());
     final EngineControlImpl engineService = new EngineControlImpl(engine);
@@ -23,7 +27,10 @@ public static void main(final String[] args) throws IOException {
             .addService(engineService)
             .build();
 
-    engine.start();
     server.start();
+
+    // In the testcontainer extension we are waiting for this line to be logged before continuing
+    // test execution. If this log gets changed please also change the extension!
+    LOG.info("ZeebeProcessTestEngine container has started ...");
   }
 }

engine-protocol/src/main/proto/engine_control.proto

@@ -65,6 +65,8 @@ service EngineControl {
     Using this while running tests concurrently will still cause issues. This is
     because the first test will be using the engine, whilst the second test will
     reset it. Therefore, the data from the first test will get lost.
+
+    After resetting the engine still needs to be started.
    */
   rpc ResetEngine (ResetEngineRequest) returns (ResetEngineResponse);
 

extension-testcontainer/src/main/java/io.camunda.zeebe.process.test.extension.testcontainer/EngineContainer.java

@@ -19,6 +19,7 @@
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.utility.DockerImageName;
 
 /** Singleton object which manages access to the testcontainer that's running the test engine */
@@ -49,6 +50,7 @@ private static void createContainer() {
         new EngineContainer(ContainerProperties.getDockerImageName())
             .withExposedPorts(
                 ContainerProperties.getContainerPort(), ContainerProperties.getGatewayPort())
-            .withLogConsumer(new Slf4jLogConsumer(LOG));
+            .withLogConsumer(new Slf4jLogConsumer(LOG))
+            .waitingFor(Wait.forLogMessage(".*ZeebeProcessTestEngine container has started.*", 1));
   }
 }

extension-testcontainer/src/main/java/io.camunda.zeebe.process.test.extension.testcontainer/ZeebeProcessTestExtension.java

@@ -44,9 +44,8 @@ public class ZeebeProcessTestExtension
   private static final String KEY_ZEEBE_ENGINE = "ZEEBE_ENGINE";
 
   /**
-   * A new testcontainer container gets created and started. This container will run the test
-   * engine. After this a {@link ContainerizedEngine} is created, which allows communicating with
-   * the testcontainer.
+   * A new testcontainer container gets created. After this a {@link ContainerizedEngine} is
+   * created, which allows communicating with the testcontainer.
    *
    * <p>Even though this method is called before each test class, the test container will only get
    * created and started once! {@link EngineContainer} is a Singleton so no new object is created
@@ -70,7 +69,7 @@ public void beforeAll(final ExtensionContext extensionContext) {
   }
 
   /**
-   * Before each test the {@link ContainerizedEngine} is reset. A client to communicate with the
+   * Before each test the {@link ContainerizedEngine} is started. A client to communicate with the
    * engine will be created, together with a {@link RecordStream}. These will be injected in the
    * fields of the test class, if they are available.
    *
@@ -80,7 +79,7 @@ public void beforeAll(final ExtensionContext extensionContext) {
   public void beforeEach(final ExtensionContext extensionContext) {
     final Object engineContent = getStore(extensionContext.getParent().get()).get(KEY_ZEEBE_ENGINE);
     final ContainerizedEngine engine = (ContainerizedEngine) engineContent;
-    engine.reset();
+    engine.start();
 
     final ZeebeClient client = engine.createClient();
     final RecordStream recordStream = RecordStream.of(new RecordStreamSourceImpl(engine));
@@ -91,7 +90,9 @@ public void beforeEach(final ExtensionContext extensionContext) {
   }
 
   /**
-   * After each test the client will be closed.
+   * After each test the client will be closed. At this point we will reset the engine. This will
+   * stop the current engine and create a new one. The new engine will not be started yet as that is
+   * done in the beforeEach method.
    *
    * @param extensionContext jUnit5 extension context
    */
@@ -100,6 +101,10 @@ public void afterEach(final ExtensionContext extensionContext) {
     final Object clientContent = getStore(extensionContext).get(KEY_ZEEBE_CLIENT);
     final ZeebeClient client = (ZeebeClient) clientContent;
     client.close();
+
+    final Object engineContent = getStore(extensionContext.getParent().get()).get(KEY_ZEEBE_ENGINE);
+    final ContainerizedEngine engine = (ContainerizedEngine) engineContent;
+    engine.reset();
   }
 
   /**
@@ -119,7 +124,7 @@ public void testFailed(final ExtensionContext extensionContext, final Throwable
   private void injectFields(final ExtensionContext extensionContext, final Object... objects) {
     final Class<?> requiredTestClass = extensionContext.getRequiredTestClass();
     final Field[] declaredFields = requiredTestClass.getDeclaredFields();
-    for (Object object : objects) {
+    for (final Object object : objects) {
       final Optional<Field> field = getField(declaredFields, object);
       field.ifPresent(value -> injectField(extensionContext, value, object));
     }
@@ -147,7 +152,7 @@ private void injectField(
     try {
       ReflectionUtils.makeAccessible(field);
       field.set(extensionContext.getRequiredTestInstance(), object);
-    } catch (IllegalAccessException e) {
+    } catch (final IllegalAccessException e) {
       throw new RuntimeException(e);
     }
   }