Repository: kroxylicious/kroxylicious (java)

Issue:
Offer record transformation convenience as a public API
As part of implementing Record Encryption we have introduced some abstractions for manipulating Records in Produce Requests and Fetch Responses while preserving batching and batch metadata.

[RecordStream](https://github.com/kroxylicious/kroxylicious/blob/a1bb61c5ec273c1154e554cef667b20318afd044/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordStream.java)
[RecordTransform](https://github.com/kroxylicious/kroxylicious/blob/main/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordTransform.java)
[BatchAwareMemoryRecordsBuilder](https://github.com/kroxylicious/kroxylicious/blob/a1bb61c5ec273c1154e554cef667b20318afd044/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/BatchAwareMemoryRecordsBuilder.java)

A single MemoryRecords of topic-partition data can be comprised of multiple batches. Each batch has some metadata. There are some special cases to consider like batches that have had 100% of their records removed and control batches that carry control messages like transaction commit/rollbacks. For the proxy to act transparently we want to ensure messages remain batched the same way and these special batches are preserved.

Making this functionality available to users as part of a public API should enable them to implement message transforms safely without having to rediscover how to work with batches (which is a bit ugly as we discovered, requiring the BatchAwareMemoryRecordsBuilder to build multi-batch MemoryRecords)

Pull request:
Extract kafka record batch transformation into re-usable module

Resolves #1127

### Type of change

- Enhancement / new feature

### Description

Creates a kroxylicious-kafka-message-tools module for classes that are purely interested in manipulating the classes in kafka-clients independent of any proxy behaviour.

This should allow user filters to use the behaviour but enable them to use a kroxylicious-kafka-message-tools version independent of the kroxylicious version. (caveat that since kroxylicious-kafka-message-tools depends on kafka-clients and kafka-clients can change, a version of kroxylicious-kafka-message-tools could become incompatible with a newer version of kroxylicious if the proxie's kafka-clients version is incompatible).

The first behaviour added to this module is record transformation. That is we want to enable filters to mutate the records contained in fetch and produce requests while preserving how the records are batched inside their MemoryRecords along with all the batch metadata.

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [ ] Write tests
- [ ] Make sure all tests pass
- [ ] Review performance test results. Ensure that any degradations to performance numbers are understood and justified.
- [ ] Make sure all Sonarcloud warnings are addressed or are justifiably ignored.
- [ ] Update documentation
- [ ] Reference relevant issue(s) and close them after merging
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


Patch:
diff --git a/kroxylicious-bom/pom.xml b/kroxylicious-bom/pom.xml
index ac40dfb6..c45d554e 100644
--- a/kroxylicious-bom/pom.xml
+++ b/kroxylicious-bom/pom.xml
@@ -141,6 +141,12 @@
                 <version>${project.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>io.kroxylicious</groupId>
+                <artifactId>kroxylicious-kafka-message-tools</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+
             <dependency>
                 <groupId>io.kroxylicious</groupId>
                 <artifactId>kroxylicious-multitenant</artifactId>
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/etc/module-layering.xml b/kroxylicious-filters/kroxylicious-record-encryption/etc/module-layering.xml
index fa7e680b..01778ecc 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/etc/module-layering.xml
+++ b/kroxylicious-filters/kroxylicious-record-encryption/etc/module-layering.xml
@@ -50,6 +50,7 @@
         <allow pkg="org.apache.kafka.common.utils"/>
         <allow pkg="org.apache.kafka.common.record"/>
         <allow pkg="org.apache.kafka.common.header"/>
+        <allow pkg="io.kroxylicious.kafka.transform"/>
         <allow pkg="io.kroxylicious.kms.service"/> <!-- serde -->
         <!-- allowed deps in this module: -->
         <allow pkg="io.kroxylicious.filter.encryption.config"/>
@@ -63,12 +64,13 @@
         <allow pkg="org.apache.kafka.common.record"/>
         <allow pkg="org.apache.kafka.common.utils"/>
         <allow pkg="org.apache.kafka.common.header"/>
+        <allow pkg="io.kroxylicious.kafka.transform"/>
         <allow pkg="com.github.benmanes.caffeine.cache"/>
         <allow pkg="io.kroxylicious.kms.service"/> <!-- serde -->
         <!-- allowed deps in this module: -->
         <allow pkg="io.kroxylicious.filter.encryption.common"/>
         <allow pkg="io.kroxylicious.filter.encryption.config"/>
-        <allow pkg="io.kroxylicious.filter.encryption.records"/>
+        <allow pkg="io.kroxylicious.kafka.transform"/>
         <allow pkg="io.kroxylicious.filter.encryption.dek"/>
         <allow pkg="io.kroxylicious.filter.encryption.crypto"/>
     </subpackage>
@@ -76,12 +78,13 @@
         <allow pkg="org.apache.kafka.common.record"/>
         <allow pkg="org.apache.kafka.common.utils"/>
         <allow pkg="org.apache.kafka.common.header"/>
+        <allow pkg="io.kroxylicious.kafka.transform"/>
         <allow pkg="com.github.benmanes.caffeine.cache"/>
         <allow pkg="io.kroxylicious.kms.service"/> <!-- serde -->
         <!-- allowed deps in this module: -->
         <allow pkg="io.kroxylicious.filter.encryption.common"/>
         <allow pkg="io.kroxylicious.filter.encryption.config"/>
-        <allow pkg="io.kroxylicious.filter.encryption.records"/>
+        <allow pkg="io.kroxylicious.kafka.transform"/>
         <allow pkg="io.kroxylicious.filter.encryption.dek"/>
         <allow pkg="io.kroxylicious.filter.encryption.crypto"/>
     </subpackage>
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/pom.xml b/kroxylicious-filters/kroxylicious-record-encryption/pom.xml
index a93c42c7..33f0795c 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/pom.xml
+++ b/kroxylicious-filters/kroxylicious-record-encryption/pom.xml
@@ -35,6 +35,10 @@
             <groupId>io.kroxylicious</groupId>
             <artifactId>kroxylicious-annotations</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-kafka-message-tools</artifactId>
+        </dependency>
         
 	    <dependency>
             <groupId>io.kroxylicious</groupId>
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/decrypt/InBandDecryptionManager.java b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/decrypt/InBandDecryptionManager.java
index 0cec301a..637d074b 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/decrypt/InBandDecryptionManager.java
+++ b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/decrypt/InBandDecryptionManager.java
@@ -29,7 +29,7 @@ import io.kroxylicious.filter.encryption.crypto.EncryptionHeader;
 import io.kroxylicious.filter.encryption.crypto.EncryptionResolver;
 import io.kroxylicious.filter.encryption.dek.Dek;
 import io.kroxylicious.filter.encryption.dek.DekManager;
-import io.kroxylicious.filter.encryption.records.RecordStream;
+import io.kroxylicious.kafka.transform.RecordStream;
 import io.kroxylicious.kms.service.Serde;
 
 import edu.umd.cs.findbugs.annotations.NonNull;
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/decrypt/RecordDecryptor.java b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/decrypt/RecordDecryptor.java
index 3414b64b..483e2676 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/decrypt/RecordDecryptor.java
+++ b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/decrypt/RecordDecryptor.java
@@ -14,7 +14,7 @@ import org.apache.kafka.common.record.Record;
 import org.apache.kafka.common.record.RecordBatch;
 
 import io.kroxylicious.filter.encryption.dek.Dek;
-import io.kroxylicious.filter.encryption.records.RecordTransform;
+import io.kroxylicious.kafka.transform.RecordTransform;
 
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/encrypt/InBandEncryptionManager.java b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/encrypt/InBandEncryptionManager.java
index a79ae1b9..da19be5f 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/encrypt/InBandEncryptionManager.java
+++ b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/encrypt/InBandEncryptionManager.java
@@ -23,7 +23,7 @@ import io.kroxylicious.filter.encryption.crypto.EncryptionHeader;
 import io.kroxylicious.filter.encryption.dek.BufferTooSmallException;
 import io.kroxylicious.filter.encryption.dek.Dek;
 import io.kroxylicious.filter.encryption.dek.ExhaustedDekException;
-import io.kroxylicious.filter.encryption.records.RecordStream;
+import io.kroxylicious.kafka.transform.RecordStream;
 import io.kroxylicious.kms.service.Serde;
 import io.kroxylicious.proxy.tag.VisibleForTesting;
 
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/encrypt/RecordEncryptor.java b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/encrypt/RecordEncryptor.java
index e6d41309..dd0cf4e0 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/encrypt/RecordEncryptor.java
+++ b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/encrypt/RecordEncryptor.java
@@ -19,7 +19,7 @@ import io.kroxylicious.filter.encryption.crypto.Encryption;
 import io.kroxylicious.filter.encryption.crypto.EncryptionHeader;
 import io.kroxylicious.filter.encryption.dek.BufferTooSmallException;
 import io.kroxylicious.filter.encryption.dek.Dek;
-import io.kroxylicious.filter.encryption.records.RecordTransform;
+import io.kroxylicious.kafka.transform.RecordTransform;
 import io.kroxylicious.kms.service.Serde;
 
 import edu.umd.cs.findbugs.annotations.NonNull;
diff --git a/kroxylicious-filters/pom.xml b/kroxylicious-filters/pom.xml
index 37be921e..1e89ce3a 100644
--- a/kroxylicious-filters/pom.xml
+++ b/kroxylicious-filters/pom.xml
@@ -54,6 +54,7 @@
                                         <include>io.kroxylicious:kroxylicious-kms-provider-kroxylicious-inmemory</include>
                                         <include>io.kroxylicious:kroxylicious-unit-test-support</include>
                                         <include>io.kroxylicious:kroxylicious-integration-test-support</include>
+                                        <include>io.kroxylicious:kroxylicious-kafka-message-tools</include>
                                     </includes>
                                 </bannedDependencies>
                             </rules>
diff --git a/kroxylicious-kafka-message-tools/etc/module-checkstyle.xml b/kroxylicious-kafka-message-tools/etc/module-checkstyle.xml
new file mode 100644
index 00000000..832f4c22
--- /dev/null
+++ b/kroxylicious-kafka-message-tools/etc/module-checkstyle.xml
@@ -0,0 +1,20 @@
+<?xml version="1.0"?>
+<!--
+
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+
+-->
+<!DOCTYPE module PUBLIC
+        "-//Checkstyle//DTD Checkstyle Configuration 1.3//EN"
+        "https://checkstyle.org/dtds/configuration_1_3.dtd">
+
+<module name="Checker">
+    <module name="TreeWalker">
+        <module name="ImportControl">
+            <property name="file" value="${project.basedir}/etc/module-layering.xml"/>
+            <property name="path" value="src/main/java" />
+        </module>
+    </module>
+</module>
diff --git a/kroxylicious-kafka-message-tools/etc/module-layering.xml b/kroxylicious-kafka-message-tools/etc/module-layering.xml
new file mode 100644
index 00000000..4c6ba670
--- /dev/null
+++ b/kroxylicious-kafka-message-tools/etc/module-layering.xml
@@ -0,0 +1,19 @@
+<?xml version="1.0"?>
+<!--
+
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+
+-->
+<!DOCTYPE import-control PUBLIC
+        "-//Checkstyle//DTD ImportControl Configuration 1.4//EN"
+        "https://checkstyle.org/dtds/import_control_1_4.dtd">
+
+<import-control pkg="io.kroxylicious.kafka">
+    <!-- Allowed globally -->
+    <allow pkg="javax?\..*" regex="true"/>
+    <allow pkg="org.slf4j"/>
+    <allow pkg="edu.umd.cs.findbugs"/>
+    <allow pkg="org.apache.kafka"/>
+</import-control>
\ No newline at end of file
diff --git a/kroxylicious-kafka-message-tools/pom.xml b/kroxylicious-kafka-message-tools/pom.xml
new file mode 100644
index 00000000..afea6b9e
--- /dev/null
+++ b/kroxylicious-kafka-message-tools/pom.xml
@@ -0,0 +1,69 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright Kroxylicious Authors.
+
+    Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>io.kroxylicious</groupId>
+        <artifactId>kroxylicious-parent</artifactId>
+        <version>0.6.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>kroxylicious-kafka-message-tools</artifactId>
+    <description>Tools for manipulating kafka messages, depends only on kafka-clients, has no dependency on the rest of Kroxylicious</description>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.github.spotbugs</groupId>
+            <artifactId>spotbugs-annotations</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-filter-test-support</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-checkstyle-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>layering</id>
+                        <goals>
+                            <goal>check</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>
\ No newline at end of file
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/BatchAwareMemoryRecordsBuilder.java b/kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/BatchAwareMemoryRecordsBuilder.java
similarity index 99%
rename from kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/BatchAwareMemoryRecordsBuilder.java
rename to kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/BatchAwareMemoryRecordsBuilder.java
index b7ba5c84..f39fad72 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/BatchAwareMemoryRecordsBuilder.java
+++ b/kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/BatchAwareMemoryRecordsBuilder.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious.filter.encryption.records;
+package io.kroxylicious.kafka.transform;
 
 import java.nio.ByteBuffer;
 import java.util.Objects;
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordConsumer.java b/kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/RecordConsumer.java
similarity index 88%
rename from kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordConsumer.java
rename to kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/RecordConsumer.java
index 0133bb10..0a47cbf2 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordConsumer.java
+++ b/kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/RecordConsumer.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious.filter.encryption.records;
+package io.kroxylicious.kafka.transform;
 
 import org.apache.kafka.common.record.Record;
 import org.apache.kafka.common.record.RecordBatch;
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordMapper.java b/kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/RecordMapper.java
similarity index 88%
rename from kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordMapper.java
rename to kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/RecordMapper.java
index 086d72ad..ca4ab4d9 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordMapper.java
+++ b/kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/RecordMapper.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious.filter.encryption.records;
+package io.kroxylicious.kafka.transform;
 
 import org.apache.kafka.common.record.Record;
 import org.apache.kafka.common.record.RecordBatch;
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordStream.java b/kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/RecordStream.java
similarity index 99%
rename from kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordStream.java
rename to kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/RecordStream.java
index 098f0140..7db55277 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordStream.java
+++ b/kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/RecordStream.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious.filter.encryption.records;
+package io.kroxylicious.kafka.transform;
 
 import java.util.ArrayList;
 import java.util.Collection;
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordTransform.java b/kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/RecordTransform.java
similarity index 98%
rename from kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordTransform.java
rename to kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/RecordTransform.java
index 26323f87..5e9cfdae 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/RecordTransform.java
+++ b/kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/RecordTransform.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious.filter.encryption.records;
+package io.kroxylicious.kafka.transform;
 
 import java.nio.ByteBuffer;
 
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/package-info.java b/kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/package-info.java
similarity index 88%
rename from kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/package-info.java
rename to kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/package-info.java
index 85bc8bf0..3e9b5763 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/records/package-info.java
+++ b/kroxylicious-kafka-message-tools/src/main/java/io/kroxylicious/kafka/transform/package-info.java
@@ -8,4 +8,4 @@
  * <p>Utilities for dealing with {@link org.apache.kafka.common.record.Record}s, {@link org.apache.kafka.common.record.RecordBatch}es
  * and {@link org.apache.kafka.common.record.MemoryRecords}.</p>
  */
-package io.kroxylicious.filter.encryption.records;
\ No newline at end of file
+package io.kroxylicious.kafka.transform;
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 7813b4ea..a7fbd67c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -362,6 +362,7 @@
 
     <modules>
         <module>kroxylicious-annotations</module>
+        <module>kroxylicious-kafka-message-tools</module>
         <module>kroxylicious-kms</module>
         <module>kroxylicious-kms-test-support</module>
         <module>kroxylicious-kms-provider-kroxylicious-inmemory</module>


Test patch:
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/common/RecordEncryptionUtilTest.java b/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/common/RecordEncryptionUtilTest.java
index 22ce089d..04b203c8 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/common/RecordEncryptionUtilTest.java
+++ b/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/common/RecordEncryptionUtilTest.java
@@ -18,7 +18,7 @@ import org.apache.kafka.common.record.TimestampType;
 import org.apache.kafka.common.utils.ByteBufferOutputStream;
 import org.junit.jupiter.api.Test;
 
-import io.kroxylicious.filter.encryption.records.BatchAwareMemoryRecordsBuilder;
+import io.kroxylicious.kafka.transform.BatchAwareMemoryRecordsBuilder;
 import io.kroxylicious.test.record.RecordTestUtils;
 
 import edu.umd.cs.findbugs.annotations.NonNull;
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/crypto/ParcelTest.java b/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/crypto/ParcelTest.java
index 7e5f3af7..471b0454 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/crypto/ParcelTest.java
+++ b/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/crypto/ParcelTest.java
@@ -32,7 +32,7 @@ import com.google.common.reflect.ClassPath;
 
 import io.kroxylicious.filter.encryption.config.ParcelVersion;
 import io.kroxylicious.filter.encryption.config.RecordField;
-import io.kroxylicious.filter.encryption.records.BatchAwareMemoryRecordsBuilder;
+import io.kroxylicious.kafka.transform.BatchAwareMemoryRecordsBuilder;
 import io.kroxylicious.test.record.RecordTestUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/encrypt/RecordEncryptorTest.java b/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/encrypt/RecordEncryptorTest.java
index 405ac383..99717b95 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/encrypt/RecordEncryptorTest.java
+++ b/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/encrypt/RecordEncryptorTest.java
@@ -30,7 +30,7 @@ import io.kroxylicious.filter.encryption.dek.Aes;
 import io.kroxylicious.filter.encryption.dek.Dek;
 import io.kroxylicious.filter.encryption.dek.DekException;
 import io.kroxylicious.filter.encryption.dek.DekManager;
-import io.kroxylicious.filter.encryption.records.RecordTransform;
+import io.kroxylicious.kafka.transform.RecordTransform;
 import io.kroxylicious.kms.service.Serde;
 import io.kroxylicious.test.assertj.KafkaAssertions;
 import io.kroxylicious.test.record.RecordTestUtils;
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/records/BatchAwareMemoryRecordsBuilderTest.java b/kroxylicious-kafka-message-tools/src/test/java/io/kroxylicious/kafka/transform/BatchAwareMemoryRecordsBuilderTest.java
similarity index 99%
rename from kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/records/BatchAwareMemoryRecordsBuilderTest.java
rename to kroxylicious-kafka-message-tools/src/test/java/io/kroxylicious/kafka/transform/BatchAwareMemoryRecordsBuilderTest.java
index 90673d9d..c4eca773 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/records/BatchAwareMemoryRecordsBuilderTest.java
+++ b/kroxylicious-kafka-message-tools/src/test/java/io/kroxylicious/kafka/transform/BatchAwareMemoryRecordsBuilderTest.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious.filter.encryption.records;
+package io.kroxylicious.kafka.transform;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/records/RecordStreamTest.java b/kroxylicious-kafka-message-tools/src/test/java/io/kroxylicious/kafka/transform/RecordStreamTest.java
similarity index 99%
rename from kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/records/RecordStreamTest.java
rename to kroxylicious-kafka-message-tools/src/test/java/io/kroxylicious/kafka/transform/RecordStreamTest.java
index 5033083a..3b5b1326 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/records/RecordStreamTest.java
+++ b/kroxylicious-kafka-message-tools/src/test/java/io/kroxylicious/kafka/transform/RecordStreamTest.java
@@ -4,7 +4,7 @@
  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
  */
 
-package io.kroxylicious.filter.encryption.records;
+package io.kroxylicious.kafka.transform;
 
 import java.nio.ByteBuffer;
 import java.nio.CharBuffer;