diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 0eb39bb..2cf9a82 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -22,10 +22,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
-      - name: Set up JDK 11
+      - name: Set up JDK 17
         uses: actions/setup-java@v2
         with:
-          java-version: '11'
+          java-version: '17'
           distribution: 'temurin'
           cache: 'maven'
       - name: Build
diff --git a/exporter/pom.xml b/exporter/pom.xml
index 33b5b5c..cd60cee 100644
--- a/exporter/pom.xml
+++ b/exporter/pom.xml
@@ -19,6 +19,17 @@
   </properties>
 
   <dependencies>
+    <!-- serialization -->
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-core</artifactId>
+    </dependency>
+
     <!-- runtime dependencies which must be packaged with the exporter -->
     <dependency>
       <groupId>io.zeebe</groupId>
@@ -43,6 +54,12 @@
       <scope>provided</scope>
     </dependency>
 
+    <dependency>
+      <groupId>io.camunda</groupId>
+      <artifactId>zeebe-protocol-jackson</artifactId>
+      <scope>provided</scope>
+    </dependency>
+
     <dependency>
       <artifactId>slf4j-api</artifactId>
       <groupId>org.slf4j</groupId>
@@ -58,13 +75,7 @@
     <!-- test dependencies -->
     <dependency>
       <groupId>io.camunda</groupId>
-      <artifactId>zeebe-protocol-jackson</artifactId>
-      <scope>test</scope>
-    </dependency>
-
-    <dependency>
-      <groupId>io.camunda</groupId>
-      <artifactId>zeebe-test</artifactId>
+      <artifactId>zeebe-exporter-test</artifactId>
       <scope>test</scope>
     </dependency>
 
diff --git a/exporter/src/main/java/io/zeebe/exporters/kafka/record/RecordSerializer.java b/exporter/src/main/java/io/zeebe/exporters/kafka/record/RecordSerializer.java
index 2b3e46a..47ddea1 100644
--- a/exporter/src/main/java/io/zeebe/exporters/kafka/record/RecordSerializer.java
+++ b/exporter/src/main/java/io/zeebe/exporters/kafka/record/RecordSerializer.java
@@ -15,8 +15,12 @@
  */
 package io.zeebe.exporters.kafka.record;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.camunda.zeebe.protocol.jackson.ZeebeProtocolModule;
 import io.camunda.zeebe.protocol.record.Record;
 import java.util.Map;
+import org.apache.kafka.common.errors.SerializationException;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 
@@ -28,13 +32,15 @@
  */
 public final class RecordSerializer implements Serializer<Record<?>> {
   private final StringSerializer delegate;
+  private final ObjectMapper mapper;
 
   public RecordSerializer() {
     this(new StringSerializer());
   }
 
   public RecordSerializer(final StringSerializer delegate) {
-    this.delegate = delegate;
+    this.mapper = new ObjectMapper().registerModule(new ZeebeProtocolModule());
+    this.delegate = new StringSerializer();
   }
 
   @Override
@@ -44,7 +50,12 @@ public void configure(final Map<String, ?> configs, final boolean isKey) {
 
   @Override
   public byte[] serialize(final String topic, final Record data) {
-    return delegate.serialize(topic, data.toJson());
+    try {
+      return delegate.serialize(topic, mapper.writeValueAsString(data));
+    } catch (JsonProcessingException e) {
+      throw new SerializationException(
+        String.format("Expected to serialize data for topic [%s], but failed", topic), e);
+    }
   }
 
   @Override
diff --git a/exporter/src/test/java/io/zeebe/exporters/kafka/KafkaExporterTest.java b/exporter/src/test/java/io/zeebe/exporters/kafka/KafkaExporterTest.java
index e426c0e..bc35333 100644
--- a/exporter/src/test/java/io/zeebe/exporters/kafka/KafkaExporterTest.java
+++ b/exporter/src/test/java/io/zeebe/exporters/kafka/KafkaExporterTest.java
@@ -19,9 +19,15 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.assertj.core.api.Assertions.tuple;
 
+import io.camunda.zeebe.exporter.test.ExporterTestConfiguration;
+import io.camunda.zeebe.exporter.test.ExporterTestContext;
+import io.camunda.zeebe.exporter.test.ExporterTestController;
+import io.camunda.zeebe.protocol.record.ImmutableRecord;
+import io.camunda.zeebe.protocol.record.Record;
+import io.camunda.zeebe.protocol.record.RecordType;
 import io.camunda.zeebe.protocol.record.ValueType;
-import io.camunda.zeebe.test.exporter.ExporterTestHarness;
-import io.camunda.zeebe.test.exporter.record.MockRecordMetadata;
+import io.camunda.zeebe.protocol.record.intent.DeploymentIntent;
+import io.camunda.zeebe.protocol.record.value.ImmutableDeploymentRecordValue;
 import io.zeebe.exporters.kafka.config.Config;
 import io.zeebe.exporters.kafka.config.parser.MockConfigParser;
 import io.zeebe.exporters.kafka.config.parser.RawConfigParser;
@@ -31,7 +37,8 @@
 import io.zeebe.exporters.kafka.producer.RecordBatchStub;
 import io.zeebe.exporters.kafka.record.RecordHandler;
 import io.zeebe.exporters.kafka.serde.RecordId;
-import java.util.stream.Collectors;
+import java.time.Duration;
+import java.util.List;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.parallel.Execution;
@@ -44,45 +51,49 @@ final class KafkaExporterTest {
 
   private final RawConfig rawConfig = new RawConfig();
   private final MockConfigParser<RawConfig, Config> mockConfigParser =
-      new MockConfigParser<>(new RawConfigParser());
+    new MockConfigParser<>(new RawConfigParser());
+
+  private final ExporterTestContext context =
+    new ExporterTestContext().setConfiguration(new ExporterTestConfiguration<>("test", rawConfig));
+  private final ExporterTestController controller = new ExporterTestController();
+
   private final RecordBatchStub.Factory batchStubFactory = new RecordBatchStub.Factory();
   private final KafkaExporter exporter = new KafkaExporter(batchStubFactory, mockConfigParser);
-  private final ExporterTestHarness testHarness = new ExporterTestHarness(exporter);
 
   @Test
-  void shouldAddRecordToBatchOnExport() throws Exception {
+  void shouldAddRecordToBatchOnExport() {
     // given
     rawConfig.maxBatchSize = 5;
-    testHarness.configure(EXPORTER_ID, rawConfig);
-    testHarness.open();
+    exporter.configure(context);
+    exporter.open(controller);
+
+    final Record<?> record = recordFixture();
 
     // when
-    final var records = testHarness.stream().export(5);
+    exporter.export(record);
 
     // then
-    final var expectedIds =
-        records.stream()
-            .map(r -> new RecordId(r.getPartitionId(), r.getPosition()))
-            .collect(Collectors.toList());
+    final var expectedIds = new RecordId(record.getPartitionId(), record.getPosition());
     assertThat(batchStubFactory.stub.getPendingRecords())
         .as("the records were added to the batch in order")
         .extracting(ProducerRecord::key)
-        .containsExactlyElementsOf(expectedIds);
+        .containsExactlyElementsOf(List.of(expectedIds));
     assertThat(batchStubFactory.stub.getFlushedRecords())
         .as("no records were flushed yet")
         .isEmpty();
   }
 
   @Test
-  void shouldUseCorrectSerializer() throws Exception {
+  void shouldUseCorrectSerializer() {
     // given
-    testHarness.configure(EXPORTER_ID, rawConfig);
-    testHarness.open();
+    exporter.configure(context);
+    exporter.open(controller);
     final var recordHandler = new RecordHandler(mockConfigParser.config.getRecords());
 
+    final Record<?> record = recordFixture();
+
     // when
-    final var json = "{\"a\": 1}";
-    final var record = testHarness.export(r -> r.setJson(json));
+    exporter.export(record);
 
     // then
     final var expectedRecord = recordHandler.transform(record);
@@ -93,18 +104,20 @@ final var record = testHarness.export(r -> r.setJson(json));
             tuple(expectedRecord.topic(), expectedRecord.key(), expectedRecord.value()));
   }
 
+
   @Test
-  void shouldSkipDisallowedRecords() throws Exception {
+  void shouldSkipDisallowedRecords() {
     // given
     rawConfig.records = new RawRecordsConfig();
     rawConfig.records.deployment = new RawRecordConfig();
     rawConfig.records.deployment.type = "";
-    testHarness.configure(EXPORTER_ID, rawConfig);
-    testHarness.open();
+    exporter.configure(context);
+    exporter.open(controller);
+
+    final Record<?> record = recordFixture();
 
     // when
-    testHarness.export(
-        r -> r.setMetadata(new MockRecordMetadata().setValueType(ValueType.DEPLOYMENT)));
+    exporter.export(record);
 
     // then
     assertThat(batchStubFactory.stub.getPendingRecords())
@@ -112,104 +125,115 @@ void shouldSkipDisallowedRecords() throws Exception {
         .isEmpty();
   }
 
+
   @Test
-  void shouldFlushOnScheduledTask() throws Exception {
+  void shouldFlushOnScheduledTask() {
     // given
-    rawConfig.maxBatchSize = 5;
-    testHarness.configure(EXPORTER_ID, rawConfig);
-    testHarness.open();
+    exporter.configure(context);
+    exporter.open(controller);
+
+    final Record<?> record = recordFixture();
 
     // when
-    final var records = testHarness.stream().export(5);
-    triggerFlushTask();
+    exporter.export(record);
+    controller.runScheduledTasks(Duration.ofSeconds(10));
 
     // then
-    final var expectedIds =
-        records.stream()
-            .map(r -> new RecordId(r.getPartitionId(), r.getPosition()))
-            .collect(Collectors.toList());
+    final var expectedIds = new RecordId(record.getPartitionId(), record.getPosition());
     assertThat(batchStubFactory.stub.getFlushedRecords())
         .as("the records were added to the batch in order")
         .extracting(ProducerRecord::key)
-        .containsExactlyElementsOf(expectedIds);
+        .containsExactlyElementsOf(List.of(expectedIds));
     assertThat(batchStubFactory.stub.getPendingRecords())
         .as("no pending records after flush")
         .isEmpty();
   }
 
   @Test
-  void shouldUpdatePositionOnFlush() throws Exception {
+  void shouldUpdatePositionOnFlush() {
     // given
-    testHarness.configure(EXPORTER_ID, rawConfig);
-    testHarness.open();
+    exporter.configure(context);
+    exporter.open(controller);
+
+    final Record<?> record = recordFixture();
 
     // when
-    final var records = testHarness.stream().export(5);
-    triggerFlushTask();
+    exporter.export(record);
+    controller.runScheduledTasks(Duration.ofSeconds(10));
 
     // then
-    assertThat(testHarness.getLastUpdatedPosition())
+    assertThat(controller.getPosition())
         .as("position should be updated since after flush")
-        .isEqualTo(records.get(4).getPosition());
+        .isEqualTo(record.getPosition());
   }
 
   @Test
-  void shouldRescheduleFlushTaskEvenOnException() throws Exception {
+  void shouldRescheduleFlushTaskEvenOnException() {
     // given
-    testHarness.configure(EXPORTER_ID, rawConfig);
-    testHarness.open();
+    exporter.configure(context);
+    exporter.open(controller);
+
+    final Record<?> record = recordFixture();
 
     // when
-    final var records = testHarness.stream().export(2);
+    exporter.export(record);
     batchStubFactory.stub.flushException = new RuntimeException("failed to flush");
-    assertThatThrownBy(this::triggerFlushTask).isEqualTo(batchStubFactory.stub.flushException);
+    assertThatThrownBy(() -> controller.runScheduledTasks(Duration.ofSeconds(10))).isEqualTo(batchStubFactory.stub.flushException);
     batchStubFactory.stub.flushException = null;
-    triggerFlushTask();
+    controller.runScheduledTasks(Duration.ofSeconds(10));
 
     // then
-    assertThat(testHarness.getLastUpdatedPosition())
+    assertThat(controller.getPosition())
         .as("position should be updated since we managed to flush after the second try")
-        .isEqualTo(records.get(1).getPosition());
+        .isEqualTo(record.getPosition());
   }
 
   @Test
-  void shouldFlushBatchOnClose() throws Exception {
+  void shouldFlushBatchOnClose() {
     // given
-    testHarness.configure(EXPORTER_ID, rawConfig);
-    testHarness.open();
+    exporter.configure(context);
+    exporter.open(controller);
+    final Record<?> record = recordFixture();
 
     // when
-    final var records = testHarness.stream().export(2);
-    testHarness.close();
+    exporter.export(record);
+    exporter.close();
 
     // then
-    assertThat(testHarness.getLastUpdatedPosition())
+    assertThat(controller.getPosition())
         .as("position should be updated since we managed to flush after the second try")
-        .isEqualTo(records.get(1).getPosition());
+        .isEqualTo(record.getPosition());
     assertThat(batchStubFactory.stub.isClosed())
         .as("batch should be closed on exporter close")
         .isTrue();
   }
 
   @Test
-  void shouldRescheduleFlush() throws Exception {
+  void shouldRescheduleFlush() {
     // given
-    testHarness.configure(EXPORTER_ID, rawConfig);
-    testHarness.open();
+    exporter.configure(context);
+    exporter.open(controller);
 
     // when
-    triggerFlushTask();
-    final var records = testHarness.stream().export(2);
-    triggerFlushTask();
+    controller.runScheduledTasks(Duration.ofSeconds(10));
+    exporter.export(recordFixture());
+    controller.runScheduledTasks(Duration.ofSeconds(10));
 
     // then
-    assertThat(testHarness.getLastUpdatedPosition())
+    assertThat(controller.getPosition())
         .as("position should be updated after triggering the second flush task")
-        .isEqualTo(records.get(1).getPosition());
+        .isEqualTo(0);
   }
 
-  private void triggerFlushTask() {
-    mockConfigParser.parse(rawConfig);
-    testHarness.runScheduledTasks(mockConfigParser.config.getFlushInterval());
+  Record<?> recordFixture() {
+    return
+      ImmutableRecord.builder()
+        .withIntent(DeploymentIntent.CREATED)
+        .withRecordType(RecordType.EVENT)
+        .withValueType(ValueType.DEPLOYMENT)
+        .withValue(ImmutableDeploymentRecordValue
+          .builder()
+          .build())
+        .build();
   }
 }
diff --git a/exporter/src/test/java/io/zeebe/exporters/kafka/config/parser/RawRecordsConfigParserTest.java b/exporter/src/test/java/io/zeebe/exporters/kafka/config/parser/RawRecordsConfigParserTest.java
index f6a665d..f6e224a 100644
--- a/exporter/src/test/java/io/zeebe/exporters/kafka/config/parser/RawRecordsConfigParserTest.java
+++ b/exporter/src/test/java/io/zeebe/exporters/kafka/config/parser/RawRecordsConfigParserTest.java
@@ -31,7 +31,15 @@
 @Execution(ExecutionMode.CONCURRENT)
 final class RawRecordsConfigParserTest {
   private static final Set<ValueType> EXPECTED_VALUE_TYPES =
-      EnumSet.complementOf(EnumSet.of(ValueType.NULL_VAL, ValueType.SBE_UNKNOWN));
+      EnumSet.complementOf(EnumSet.of(
+        ValueType.NULL_VAL,
+        ValueType.SBE_UNKNOWN,
+        ValueType.DECISION,
+        ValueType.DECISION_REQUIREMENTS,
+        ValueType.DECISION_EVALUATION,
+        ValueType.CHECKPOINT,
+        ValueType.PROCESS_INSTANCE_MODIFICATION
+      ));
 
   private final RawRecordsConfigParser parser = new RawRecordsConfigParser();
 
diff --git a/exporter/src/test/java/io/zeebe/exporters/kafka/record/RecordHandlerTest.java b/exporter/src/test/java/io/zeebe/exporters/kafka/record/RecordHandlerTest.java
index 57ee520..943118e 100644
--- a/exporter/src/test/java/io/zeebe/exporters/kafka/record/RecordHandlerTest.java
+++ b/exporter/src/test/java/io/zeebe/exporters/kafka/record/RecordHandlerTest.java
@@ -17,17 +17,19 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-import io.camunda.zeebe.protocol.jackson.record.DeploymentRecordValueBuilder;
-import io.camunda.zeebe.protocol.jackson.record.RecordBuilder;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.camunda.zeebe.protocol.jackson.ZeebeProtocolModule;
+import io.camunda.zeebe.protocol.record.ImmutableRecord;
 import io.camunda.zeebe.protocol.record.Record;
 import io.camunda.zeebe.protocol.record.RecordType;
 import io.camunda.zeebe.protocol.record.ValueType;
 import io.camunda.zeebe.protocol.record.intent.DeploymentIntent;
-import io.camunda.zeebe.protocol.record.value.DeploymentRecordValue;
+import io.camunda.zeebe.protocol.record.value.ImmutableDeploymentRecordValue;
 import io.zeebe.exporters.kafka.config.RecordConfig;
 import io.zeebe.exporters.kafka.config.RecordsConfig;
 import io.zeebe.exporters.kafka.serde.RecordId;
-import java.nio.charset.StandardCharsets;
 import java.util.EnumSet;
 import java.util.Map;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -42,10 +44,10 @@ final class RecordHandlerTest {
       new RecordConfig(EnumSet.allOf(RecordType.class), "zeebe");
 
   @Test
-  void shouldTransformRecord() {
+  void shouldTransformRecord() throws JsonProcessingException {
     // given
-    final Record<DeploymentRecordValue> record =
-        buildDeploymentRecord().recordType(RecordType.COMMAND).build();
+    final ObjectMapper mapper = new ObjectMapper().registerModule(new ZeebeProtocolModule());
+    final Record<?> record = buildDeploymentRecord().withRecordType(RecordType.COMMAND).build();
     final RecordConfig deploymentRecordConfig =
         new RecordConfig(EnumSet.allOf(RecordType.class), "topic");
     final RecordHandler recordHandler = new RecordHandler(newRecordsConfig(RecordType.COMMAND));
@@ -57,14 +59,13 @@ void shouldTransformRecord() {
     assertThat(transformed.topic()).isEqualTo(deploymentRecordConfig.getTopic());
     assertThat(transformed.key())
         .isEqualTo(new RecordId(record.getPartitionId(), record.getPosition()));
-    assertThat(transformed.value()).isEqualTo(record.toJson().getBytes(StandardCharsets.UTF_8));
+    assertThat(transformed.value()).isEqualTo(mapper.writeValueAsBytes(record));
   }
 
   @Test
   void shouldTestRecordAsNotAllowed() {
     // given
-    final Record<DeploymentRecordValue> record =
-        buildDeploymentRecord().recordType(RecordType.COMMAND).build();
+    final Record<?> record = buildDeploymentRecord().withRecordType(RecordType.COMMAND).build();
     final RecordHandler recordHandler = new RecordHandler(newRecordsConfig(RecordType.EVENT));
 
     // when - then
@@ -74,8 +75,7 @@ void shouldTestRecordAsNotAllowed() {
   @Test
   void shouldTestRecordAsAllowed() {
     // given
-    final Record<DeploymentRecordValue> record =
-        buildDeploymentRecord().recordType(RecordType.EVENT).build();
+    final Record<?> record = buildDeploymentRecord().withRecordType(RecordType.EVENT).build();
     final RecordHandler recordHandler = new RecordHandler(newRecordsConfig(RecordType.EVENT));
 
     // when - then
@@ -87,14 +87,15 @@ private RecordsConfig newRecordsConfig(final RecordType allowedType) {
     return new RecordsConfig(Map.of(ValueType.DEPLOYMENT, recordConfig), DEFAULT_RECORD_CONFIG);
   }
 
-  private RecordBuilder<DeploymentRecordValue> buildDeploymentRecord() {
-    return new RecordBuilder<DeploymentRecordValue>()
-        .valueType(ValueType.DEPLOYMENT)
-        .recordType(RecordType.EVENT)
-        .timestamp(System.currentTimeMillis())
-        .intent(DeploymentIntent.CREATE)
-        .value(new DeploymentRecordValueBuilder().build())
-        .partitionId(1)
-        .position(1);
+  private ImmutableRecord.Builder<?> buildDeploymentRecord() {
+    return ImmutableRecord.builder()
+      .withIntent(DeploymentIntent.CREATE)
+      .withRecordType(RecordType.EVENT)
+      .withValueType(ValueType.DEPLOYMENT)
+      .withPartitionId(1)
+      .withPosition(1)
+      .withValue(ImmutableDeploymentRecordValue
+        .builder()
+        .build());
   }
 }
diff --git a/pom.xml b/pom.xml
index 5d74229..9195292 100644
--- a/pom.xml
+++ b/pom.xml
@@ -26,7 +26,7 @@
 
   <properties>
     <!-- release parent settings -->
-    <version.java>11</version.java>
+    <version.java>17</version.java>
     <nexus.snapshot.repository>
       https://artifacts.camunda.com/artifactory/zeebe-io-snapshots/
     </nexus.snapshot.repository>
@@ -51,8 +51,8 @@
     <version.slf4j>1.7.36</version.slf4j>
     <version.spotbugs>4.2.0</version.spotbugs>
     <version.testcontainers>1.17.4</version.testcontainers>
-    <version.zeebe>1.3.6</version.zeebe>
     <version.zeebe-test-container>3.5.2</version.zeebe-test-container>
+    <version.zeebe>8.1.2</version.zeebe>
 
     <!-- plugin version -->
     <plugin.version.animal-sniffer>1.22</plugin.version.animal-sniffer>
@@ -60,11 +60,10 @@
     <plugin.version.compiler>3.10.1</plugin.version.compiler>
     <plugin.version.checkstyle>3.2.0</plugin.version.checkstyle>
     <plugin.version.clean>3.2.0</plugin.version.clean>
-    <plugin.version.dependency-analyzer>1.11.1</plugin.version.dependency-analyzer>
-    <plugin.version.dependency>3.1.2</plugin.version.dependency>
+    <plugin.version.dependency-analyzer>1.11.2</plugin.version.dependency-analyzer>
+    <plugin.version.dependency>3.3.0</plugin.version.dependency>
     <plugin.version.enforcer>3.1.0</plugin.version.enforcer>
     <plugin.version.failsafe>3.0.0-M7</plugin.version.failsafe>
-    <plugin.version.fmt>2.13</plugin.version.fmt>
     <plugin.version.gpg>3.0.1</plugin.version.gpg>
     <plugin.version.javadoc>3.4.1</plugin.version.javadoc>
     <plugin.version.license>4.1</plugin.version.license>
@@ -72,6 +71,7 @@
     <plugin.version.proto-compat>1.0.6</plugin.version.proto-compat>
     <plugin.version.revapi>0.14.7</plugin.version.revapi>
     <plugin.version.surefire>3.0.0-M7</plugin.version.surefire>
+    <plugin.version.spotless>2.17.3</plugin.version.spotless>
 
     <!-- maven extensions -->
     <extension.version.os-maven-plugin>1.6.1</extension.version.os-maven-plugin>
@@ -88,6 +88,7 @@
     <enforcer.skip>${skipChecks}</enforcer.skip>
     <mdep.analyze.skip>${skipChecks}</mdep.analyze.skip>
     <revapi.skip>${skipChecks}</revapi.skip>
+    <spotless.apply.skip>${skipChecks}</spotless.apply.skip>
     <spotless.check.skip>${skipChecks}</spotless.check.skip>
   </properties>
 
@@ -115,6 +116,24 @@
         <type>pom</type>
       </dependency>
 
+      <dependency>
+        <groupId>io.camunda</groupId>
+        <artifactId>zeebe-protocol-jackson</artifactId>
+        <version>${version.zeebe}</version>
+      </dependency>
+
+      <dependency>
+        <groupId>io.camunda</groupId>
+        <artifactId>zeebe-protocol-asserts</artifactId>
+        <version>${version.zeebe}</version>
+      </dependency>
+
+      <dependency>
+        <groupId>io.camunda</groupId>
+        <artifactId>zeebe-exporter-test</artifactId>
+        <version>${version.zeebe}</version>
+      </dependency>
+
       <!-- Integration testing via Docker -->
       <dependency>
         <groupId>org.testcontainers</groupId>
@@ -130,6 +149,12 @@
         <version>${version.zeebe-test-container}</version>
       </dependency>
 
+      <dependency>
+        <groupId>io.camunda</groupId>
+        <artifactId>zeebe-process-test-extension</artifactId>
+        <version>${version.zeebe}</version>
+      </dependency>
+
       <!-- Unit test engine -->
       <dependency>
         <groupId>org.junit</groupId>
@@ -351,16 +376,17 @@
 
         <!-- Google code format plugin -->
         <plugin>
-          <groupId>com.coveo</groupId>
-          <artifactId>fmt-maven-plugin</artifactId>
-          <version>${plugin.version.fmt}</version>
-          <executions>
-            <execution>
-              <goals>
-                <goal>format</goal>
-              </goals>
-            </execution>
-          </executions>
+          <groupId>com.diffplug.spotless</groupId>
+          <artifactId>spotless-maven-plugin</artifactId>
+          <version>${plugin.version.spotless}</version>
+          <configuration>
+            <java>
+              <googleJavaFormat>
+                <version>1.12.0</version>
+                <style>GOOGLE</style>
+              </googleJavaFormat>
+            </java>
+          </configuration>
         </plugin>
 
         <!-- jar with dependency assembly -->
@@ -403,13 +429,6 @@
           <artifactId>maven-dependency-plugin</artifactId>
           <version>${plugin.version.dependency}</version>
           <!-- To run with Java 11 - https://issues.apache.org/jira/browse/MDEP-613-->
-          <dependencies>
-            <dependency>
-              <groupId>org.apache.maven.shared</groupId>
-              <artifactId>maven-dependency-analyzer</artifactId>
-              <version>${plugin.version.dependency-analyzer}</version>
-            </dependency>
-          </dependencies>
 
           <executions>
             <execution>
@@ -497,11 +516,6 @@
         <artifactId>maven-surefire-plugin</artifactId>
       </plugin>
 
-      <plugin>
-        <groupId>com.coveo</groupId>
-        <artifactId>fmt-maven-plugin</artifactId>
-      </plugin>
-
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-enforcer-plugin</artifactId>
diff --git a/qa/pom.xml b/qa/pom.xml
index fcc1574..56fb3df 100644
--- a/qa/pom.xml
+++ b/qa/pom.xml
@@ -29,12 +29,6 @@
       <scope>test</scope>
     </dependency>
 
-    <dependency>
-      <groupId>io.camunda</groupId>
-      <artifactId>zeebe-protocol-jackson</artifactId>
-      <scope>test</scope>
-    </dependency>
-
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-core</artifactId>
diff --git a/qa/src/test/java/io/zeebe/exporters/kafka/qa/DebugHttpExporterClient.java b/qa/src/test/java/io/zeebe/exporters/kafka/qa/DebugHttpExporterClient.java
index ce502a9..84f4a94 100644
--- a/qa/src/test/java/io/zeebe/exporters/kafka/qa/DebugHttpExporterClient.java
+++ b/qa/src/test/java/io/zeebe/exporters/kafka/qa/DebugHttpExporterClient.java
@@ -18,7 +18,6 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectReader;
-import io.camunda.zeebe.protocol.jackson.record.AbstractRecord;
 import io.camunda.zeebe.protocol.record.Record;
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -42,7 +41,7 @@
 final class DebugHttpExporterClient {
 
   private static final ObjectReader READER =
-      new ObjectMapper().readerFor(new TypeReference<List<AbstractRecord<?>>>() {});
+      new ObjectMapper().readerFor(new TypeReference<List<Record<?>>>() {});
 
   private final URL serverUrl;
 
@@ -53,7 +52,7 @@ final class DebugHttpExporterClient {
   Stream<Record<?>> streamRecords() {
     try {
       // the HTTP exporter returns records in reversed order, so flip them before returning
-      final List<AbstractRecord<?>> records = READER.readValue(serverUrl);
+      final List<Record<?>> records = READER.readValue(serverUrl);
       Collections.reverse(records);
 
       return records.stream().map(r -> r);
diff --git a/serde/pom.xml b/serde/pom.xml
index 469596d..88d08a0 100644
--- a/serde/pom.xml
+++ b/serde/pom.xml
@@ -22,11 +22,6 @@
 
   <dependencies>
     <!-- serialization -->
-    <dependency>
-      <groupId>io.camunda</groupId>
-      <artifactId>zeebe-protocol-jackson</artifactId>
-    </dependency>
-
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-databind</artifactId>
@@ -49,6 +44,11 @@
     </dependency>
 
     <!-- Zeebe -->
+    <dependency>
+      <groupId>io.camunda</groupId>
+      <artifactId>zeebe-protocol-jackson</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>io.camunda</groupId>
       <artifactId>zeebe-protocol</artifactId>
diff --git a/serde/src/main/java/io/zeebe/exporters/kafka/serde/RecordDeserializer.java b/serde/src/main/java/io/zeebe/exporters/kafka/serde/RecordDeserializer.java
index 76bb524..93af2aa 100644
--- a/serde/src/main/java/io/zeebe/exporters/kafka/serde/RecordDeserializer.java
+++ b/serde/src/main/java/io/zeebe/exporters/kafka/serde/RecordDeserializer.java
@@ -18,7 +18,7 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectReader;
-import io.camunda.zeebe.protocol.jackson.record.AbstractRecord;
+import io.camunda.zeebe.protocol.jackson.ZeebeProtocolModule;
 import io.camunda.zeebe.protocol.record.Record;
 import org.apache.kafka.common.serialization.Deserializer;
 
@@ -31,11 +31,11 @@
 public final class RecordDeserializer extends JacksonDeserializer<Record<?>> {
 
   public RecordDeserializer() {
-    this(new ObjectMapper());
+    this(new ObjectMapper().registerModule(new ZeebeProtocolModule()));
   }
 
   public RecordDeserializer(final ObjectMapper objectMapper) {
-    this(objectMapper.readerFor(new TypeReference<AbstractRecord<?>>() {}));
+    this(objectMapper.registerModule(new ZeebeProtocolModule()).readerFor(new TypeReference<Record<?>>() {}));
   }
 
   public RecordDeserializer(final ObjectReader objectReader) {
diff --git a/serde/src/main/java/io/zeebe/exporters/kafka/serde/RecordSerializer.java b/serde/src/main/java/io/zeebe/exporters/kafka/serde/RecordSerializer.java
index 6ccf097..93865dd 100644
--- a/serde/src/main/java/io/zeebe/exporters/kafka/serde/RecordSerializer.java
+++ b/serde/src/main/java/io/zeebe/exporters/kafka/serde/RecordSerializer.java
@@ -18,7 +18,7 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectWriter;
-import io.camunda.zeebe.protocol.jackson.record.AbstractRecord;
+import io.camunda.zeebe.protocol.jackson.ZeebeProtocolModule;
 import io.camunda.zeebe.protocol.record.Record;
 import org.apache.kafka.common.serialization.Serializer;
 
@@ -34,11 +34,11 @@
  */
 public final class RecordSerializer extends JacksonSerializer<Record<?>> {
   public RecordSerializer() {
-    this(new ObjectMapper());
+    this(new ObjectMapper().registerModule(new ZeebeProtocolModule()));
   }
 
   protected RecordSerializer(final ObjectMapper objectMapper) {
-    this(objectMapper.writerFor(new TypeReference<AbstractRecord<?>>() {}));
+    this(objectMapper.registerModule(new ZeebeProtocolModule()).writerFor(new TypeReference<Record<?>>() {}));
   }
 
   protected RecordSerializer(final ObjectWriter writer) {
diff --git a/serde/src/test/java/io/zeebe/exporters/kafka/serde/RecordTest.java b/serde/src/test/java/io/zeebe/exporters/kafka/serde/RecordTest.java
index efb671d..9dcf6a9 100644
--- a/serde/src/test/java/io/zeebe/exporters/kafka/serde/RecordTest.java
+++ b/serde/src/test/java/io/zeebe/exporters/kafka/serde/RecordTest.java
@@ -19,13 +19,12 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.cbor.databind.CBORMapper;
-import io.camunda.zeebe.protocol.jackson.record.DeploymentRecordValueBuilder;
-import io.camunda.zeebe.protocol.jackson.record.RecordBuilder;
+import io.camunda.zeebe.protocol.record.ImmutableRecord;
 import io.camunda.zeebe.protocol.record.Record;
 import io.camunda.zeebe.protocol.record.RecordType;
 import io.camunda.zeebe.protocol.record.ValueType;
 import io.camunda.zeebe.protocol.record.intent.DeploymentIntent;
-import io.camunda.zeebe.protocol.record.value.DeploymentRecordValue;
+import io.camunda.zeebe.protocol.record.value.ImmutableDeploymentRecordValue;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.parallel.Execution;
 import org.junit.jupiter.api.parallel.ExecutionMode;
@@ -37,13 +36,15 @@ final class RecordTest {
   @Test
   void shouldSerialize() {
     // given
+    ImmutableDeploymentRecordValue.builder().build();
     final Record<?> record =
-        new RecordBuilder<DeploymentRecordValue>()
-            .intent(DeploymentIntent.CREATED)
-            .recordType(RecordType.EVENT)
-            .valueType(ValueType.DEPLOYMENT)
-            .value(new DeploymentRecordValueBuilder().build())
-            .build();
+        ImmutableRecord.builder()
+          .withIntent(DeploymentIntent.CREATED)
+          .withRecordType(RecordType.EVENT)
+          .withValueType(ValueType.DEPLOYMENT)
+          .withValue(ImmutableDeploymentRecordValue.builder().build())
+          .build();
+
     final RecordSerializer serializer = new RecordSerializer();
     final RecordDeserializer deserializer = new RecordDeserializer();
 
@@ -62,12 +63,12 @@ void shouldSerializeOtherFormat() {
     // given
     final ObjectMapper cborMapper = new CBORMapper();
     final Record<?> record =
-        new RecordBuilder<DeploymentRecordValue>()
-            .intent(DeploymentIntent.CREATED)
-            .recordType(RecordType.EVENT)
-            .valueType(ValueType.DEPLOYMENT)
-            .value(new DeploymentRecordValueBuilder().build())
-            .build();
+      ImmutableRecord.builder()
+        .withIntent(DeploymentIntent.CREATED)
+        .withRecordType(RecordType.EVENT)
+        .withValueType(ValueType.DEPLOYMENT)
+        .withValue(ImmutableDeploymentRecordValue.builder().build())
+        .build();
     final RecordSerializer serializer = new RecordSerializer(cborMapper);
     final RecordDeserializer deserializer = new RecordDeserializer(cborMapper);