test: only use form linking only to produce bug

The problem is not related to form deploy, but to filling an empty cache
on form linking.

Rather than deploying a new form after the restart, we can simply ensure
we have 2 forms available to link. Then when forms are being linked
after snapshot and restart the bug surfaces.

(cherry picked from commit a35228c)

qa/integration-tests/src/test/java/io/camunda/zeebe/it/processing/FormLinkingIT.java

@@ -12,7 +12,6 @@
 import io.camunda.zeebe.client.ZeebeClient;
 import io.camunda.zeebe.client.api.response.DeploymentEvent;
 import io.camunda.zeebe.model.bpmn.Bpmn;
-import io.camunda.zeebe.protocol.record.intent.FormIntent;
 import io.camunda.zeebe.protocol.record.intent.JobIntent;
 import io.camunda.zeebe.qa.util.actuator.PartitionsActuator;
 import io.camunda.zeebe.qa.util.cluster.TestStandaloneBroker;
@@ -21,12 +20,12 @@
 import io.camunda.zeebe.snapshots.impl.FileBasedSnapshotId;
 import io.camunda.zeebe.test.util.junit.AutoCloseResources;
 import io.camunda.zeebe.test.util.junit.AutoCloseResources.AutoCloseResource;
+import io.camunda.zeebe.test.util.junit.RegressionTest;
 import io.camunda.zeebe.test.util.record.RecordingExporter;
 import java.time.Duration;
 import java.util.Optional;
 import org.awaitility.Awaitility;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
 
 @AutoCloseResources
 @ZeebeIntegration
@@ -41,11 +40,9 @@ void beforeEach() {
     client = zeebe.newClientBuilder().build();
   }
 
-  @Test
+  @RegressionTest("https://github.com/camunda/zeebe/issues/16311")
   public void shouldActivateUserTaskWithCorrectFormKey() {
     // given
-    final String form1Path = "form/form-linking-test-form-1.form";
-
     final DeploymentEvent deployment =
         client
             .newDeployResourceCommand()
@@ -57,11 +54,21 @@ public void shouldActivateUserTaskWithCorrectFormKey() {
                     .endEvent()
                     .done(),
                 "form_linking_test.bpmn")
-            .addResourceFromClasspath(form1Path)
+            .addProcessModel(
+                Bpmn.createExecutableProcess("form_linking_test2")
+                    .startEvent()
+                    .userTask()
+                    .zeebeFormId("formId2")
+                    .endEvent()
+                    .done(),
+                "form_linking_test2.bpmn")
+            .addResourceFromClasspath("form/form-linking-test-form-1.form")
+            .addResourceFromClasspath("form/form-linking-test-form-2.form")
             .send()
             .join();
 
-    final Long formKey = deployment.getForm().getFirst().getFormKey();
+    final var formKey = deployment.getForm().getFirst().getFormKey();
+    final var formKey2 = deployment.getForm().getLast().getFormKey();
 
     // take snapshot and start the engine from snapshot
     partitions.takeSnapshot();
@@ -78,46 +85,33 @@ public void shouldActivateUserTaskWithCorrectFormKey() {
     // when
     zeebe.withRecordingExporter(true).start().awaitCompleteTopology();
 
-    // create a process instance to trigger form linking where the actual caching issue exists
-    final long processInstanceKey =
-        client
-            .newCreateInstanceCommand()
-            .bpmnProcessId("form_linking_test")
-            .latestVersion()
-            .send()
-            .join()
-            .getProcessInstanceKey();
-    assertThat(
-            RecordingExporter.jobRecords(JobIntent.CREATED)
-                .withProcessInstanceKey(processInstanceKey)
-                .getFirst()
-                .getValue()
-                .getCustomHeaders())
-        .containsValue(formKey.toString());
+    // then
 
-    // deploy another form to update form object referenced by formId1 in the cache (the wrong
-    // behaviour)
-    final String form2Path = "form/form-linking-test-form-2.form";
-    client.newDeployResourceCommand().addResourceFromClasspath(form2Path).send().join();
-    RecordingExporter.formRecords().withIntent(FormIntent.CREATED).withFormId("formId2").await();
+    // fills the cache correctly
+    assertCorrectFormLinkedForProcess("form_linking_test", formKey);
 
-    // create another process instance to verify it works as expected (e.g. cache data is not
-    // changed)
-    final long processInstanceKey2 =
+    // previously corrupted the cache
+    assertCorrectFormLinkedForProcess("form_linking_test2", formKey2);
+
+    // failed previously because of corrupted cache
+    assertCorrectFormLinkedForProcess("form_linking_test", formKey);
+  }
+
+  private void assertCorrectFormLinkedForProcess(final String processId, final long formKey) {
+    final long processInstanceKey =
         client
             .newCreateInstanceCommand()
-            .bpmnProcessId("form_linking_test")
+            .bpmnProcessId(processId)
             .latestVersion()
             .send()
             .join()
             .getProcessInstanceKey();
-
     assertThat(
             RecordingExporter.jobRecords(JobIntent.CREATED)
-                .withProcessInstanceKey(processInstanceKey2)
+                .withProcessInstanceKey(processInstanceKey)
                 .getFirst()
                 .getValue()
                 .getCustomHeaders())
-        .containsValue(formKey.toString());
+        .containsValue(String.valueOf(formKey));
   }
 }