diff --git a/src/main/java/io/debezium/connector/vitess/converters/VitessCloudEventsProvider.java b/src/main/java/io/debezium/connector/vitess/converters/VitessCloudEventsProvider.java
index 8db741b3..c5649681 100644
--- a/src/main/java/io/debezium/connector/vitess/converters/VitessCloudEventsProvider.java
+++ b/src/main/java/io/debezium/connector/vitess/converters/VitessCloudEventsProvider.java
@@ -5,10 +5,8 @@
  */
 package io.debezium.connector.vitess.converters;
 
-import org.apache.kafka.connect.data.Schema;
-import org.apache.kafka.connect.data.Struct;
-
 import io.debezium.connector.vitess.Module;
+import io.debezium.converters.recordandmetadata.RecordAndMetadata;
 import io.debezium.converters.spi.CloudEventsMaker;
 import io.debezium.converters.spi.CloudEventsProvider;
 import io.debezium.converters.spi.RecordParser;
@@ -26,8 +24,8 @@ public String getName() {
     }
 
     @Override
-    public RecordParser createParser(Schema schema, Struct record) {
-        return new VitessRecordParser(schema, record);
+    public RecordParser createParser(RecordAndMetadata recordAndMetadata) {
+        return new VitessRecordParser(recordAndMetadata);
     }
 
     @Override
diff --git a/src/main/java/io/debezium/connector/vitess/converters/VitessRecordParser.java b/src/main/java/io/debezium/connector/vitess/converters/VitessRecordParser.java
index 775c13ef..8d16318f 100644
--- a/src/main/java/io/debezium/connector/vitess/converters/VitessRecordParser.java
+++ b/src/main/java/io/debezium/connector/vitess/converters/VitessRecordParser.java
@@ -7,11 +7,10 @@
 
 import java.util.Set;
 
-import org.apache.kafka.connect.data.Schema;
-import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.errors.DataException;
 
 import io.debezium.connector.vitess.SourceInfo;
+import io.debezium.converters.recordandmetadata.RecordAndMetadata;
 import io.debezium.converters.spi.RecordParser;
 import io.debezium.data.Envelope;
 import io.debezium.util.Collect;
@@ -25,8 +24,8 @@ public class VitessRecordParser extends RecordParser {
 
     private static final Set<String> VITESS_SOURCE_FIELD = Collect.unmodifiableSet(SourceInfo.VGTID_KEY, SourceInfo.KEYSPACE_NAME_KEY);
 
-    public VitessRecordParser(Schema schema, Struct record) {
-        super(schema, record, Envelope.FieldName.BEFORE, Envelope.FieldName.AFTER);
+    public VitessRecordParser(RecordAndMetadata recordAndMetadata) {
+        super(recordAndMetadata, Envelope.FieldName.BEFORE, Envelope.FieldName.AFTER);
     }
 
     @Override