merge: #15201

15201: Replay deployed forms r=nicpuppa a=korthout

## Description

<!-- Please explain the changes you made here. -->

Includes `Form` records in the supported Engine records so they are replayed.

A test case is added as a regression test.

## Related issues

<!-- Which issues are closed by this PR or are related -->

closes #15194



Co-authored-by: Nico Korthout <nico.korthout@camunda.com>

engine/src/main/java/io/camunda/zeebe/engine/Engine.java

@@ -46,7 +46,7 @@ public class Engine implements RecordProcessor {
       "Expected to process record '%s' without errors, but exception occurred with message '%s'.";
 
   private static final EnumSet<ValueType> SUPPORTED_VALUETYPES =
-      EnumSet.range(ValueType.JOB, ValueType.MESSAGE_BATCH);
+      EnumSet.range(ValueType.JOB, ValueType.FORM);
 
   private EventApplier eventApplier;
   private RecordProcessorMap recordProcessorMap;

engine/src/test/java/io/camunda/zeebe/engine/state/ReplayStateTest.java

@@ -20,6 +20,7 @@
 import io.camunda.zeebe.protocol.record.intent.ProcessInstanceIntent;
 import io.camunda.zeebe.protocol.record.value.BpmnElementType;
 import io.camunda.zeebe.protocol.record.value.JobBatchRecordValue;
+import io.camunda.zeebe.protocol.record.value.JobRecordValue;
 import io.camunda.zeebe.stream.impl.StreamProcessor.Phase;
 import io.camunda.zeebe.test.util.record.RecordingExporter;
 import java.time.Duration;
@@ -312,6 +313,23 @@ public static Collection<TestCase> testRecords() {
                       .withElementType(BpmnElementType.PROCESS)
                       .getFirst();
                 }),
+        testCase("link deployed form")
+            .withProcess(
+                Bpmn.createExecutableProcess(PROCESS_ID)
+                    .startEvent()
+                    .userTask("task", t -> t.zeebeFormId("Form_0w7r08e"))
+                    .endEvent()
+                    .done())
+            .withForm("/form/test-form-1.form")
+            .withExecution(
+                engine -> {
+                  engine.processInstance().ofBpmnProcessId(PROCESS_ID).create();
+                  final Record<JobRecordValue> job =
+                      RecordingExporter.jobRecords(JobIntent.CREATED).getFirst();
+                  assertThat(job.getValue().getCustomHeaders())
+                      .containsKey("io.camunda.zeebe:formKey");
+                  return job;
+                }),
         testCase("correlate buffered message to start event")
             .withProcess(
                 Bpmn.createExecutableProcess(PROCESS_ID)
@@ -357,6 +375,7 @@ public void init() {
   public void shouldRestoreState() {
     // given
     testCase.processes.forEach(process -> engine.deployment().withXmlResource(process).deploy());
+    testCase.forms.forEach(form -> engine.deployment().withJsonClasspathResource(form).deploy());
 
     final Record<?> finalRecord = testCase.execution.apply(engine);
 
@@ -421,6 +440,7 @@ private static TestCase testCase(final String description) {
   private static final class TestCase {
     private final String description;
     private final List<BpmnModelInstance> processes = new ArrayList<>();
+    private final List<String> forms = new ArrayList<>();
     private Function<EngineRule, Record<?>> execution =
         engine -> RecordingExporter.records().getFirst();
 
@@ -433,6 +453,11 @@ private TestCase withProcess(final BpmnModelInstance process) {
       return this;
     }
 
+    public TestCase withForm(final String pathToForm) {
+      forms.add(pathToForm);
+      return this;
+    }
+
     private TestCase withExecution(final Function<EngineRule, Record<?>> execution) {
       this.execution = execution;
       return this;