diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index f4c1fdc..fc4b52e 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -18,8 +18,7 @@ jobs:
       max-parallel: 100
       matrix:
         spring_boot_version:
-          - 3.4.6
-          - 3.5.0
+          - 4.0.6
     env:
       SPRING_BOOT_VERSION: ${{ matrix.spring_boot_version }}
       GRADLE_OPTS: "-Djava.security.egd=file:/dev/./urandom -Dorg.gradle.parallel=true"
@@ -52,29 +51,35 @@ jobs:
           ALLOW_ANONYMOUS_LOGIN: "yes"
           JVMFLAGS: "-Xmx512m -Xms64m"
       kafka1:
-        image: wurstmeister/kafka:2.12-2.2.0
+        image: bitnamilegacy/kafka:3.6.0
         env:
-          KAFKA_BROKER_ID: 1
-          KAFKA_ZOOKEEPER_CONNECT: zookeeper1:2181/kafka1
-          KAFKA_LISTENERS: PLAINTEXT://:9092
-          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-          KAFKA_MESSAGE_MAX_BYTES: "10485760"
-          KAFKA_ZOOKEEPER_SESSION_TIMEOUT_MS: 20000
-          KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE: "true"
-          KAFKA_LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS: 5
-          KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
+          ALLOW_PLAINTEXT_LISTENER: "yes"
+          KAFKA_CFG_BROKER_ID: 1
+          KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper1:2181/kafka1
+          KAFKA_CFG_LISTENERS: PLAINTEXT://:9092
+          KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092
+          KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+          KAFKA_CFG_MESSAGE_MAX_BYTES: "10485760"
+          KAFKA_CFG_ZOOKEEPER_SESSION_TIMEOUT_MS: 20000
+          KAFKA_CFG_UNCLEAN_LEADER_ELECTION_ENABLE: "true"
+          KAFKA_CFG_LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS: 5
+          KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "false"
+          KAFKA_HEAP_OPTS: -Xms256m -Xmx768m
       kafka2:
-        image: wurstmeister/kafka:2.12-2.2.0
+        image: bitnamilegacy/kafka:3.6.0
         env:
-          KAFKA_BROKER_ID: 1
-          KAFKA_ZOOKEEPER_CONNECT: zookeeper1:2181/kafka2
-          KAFKA_LISTENERS: PLAINTEXT://:9092
-          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-          KAFKA_MESSAGE_MAX_BYTES: "10485760"
-          KAFKA_ZOOKEEPER_SESSION_TIMEOUT_MS: 20000
-          KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE: "true"
-          KAFKA_LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS: 5
-          KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
+          ALLOW_PLAINTEXT_LISTENER: "yes"
+          KAFKA_CFG_BROKER_ID: 1
+          KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper1:2181/kafka2
+          KAFKA_CFG_LISTENERS: PLAINTEXT://:9092
+          KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka2:9092
+          KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+          KAFKA_CFG_MESSAGE_MAX_BYTES: "10485760"
+          KAFKA_CFG_ZOOKEEPER_SESSION_TIMEOUT_MS: 20000
+          KAFKA_CFG_UNCLEAN_LEADER_ELECTION_ENABLE: "true"
+          KAFKA_CFG_LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS: 5
+          KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "false"
+          KAFKA_HEAP_OPTS: -Xms256m -Xmx768m
     steps:
       - name: "Install packages"
         # We do some changes as the main ubuntu repo is currently very slow.
diff --git a/build.common.gradle b/build.common.gradle
index 383d0d5..7234a25 100644
--- a/build.common.gradle
+++ b/build.common.gradle
@@ -75,7 +75,7 @@ dependencies {
 }
 
 java {
-    if (springBootVersion.startsWith("3.")) {
+    if (springBootVersion.startsWith("3.") || springBootVersion.startsWith("4.")) {
         sourceCompatibility = JavaVersion.VERSION_17
         targetCompatibility = JavaVersion.VERSION_17
     } else {
@@ -84,6 +84,10 @@ java {
     }
     withSourcesJar()
     withJavadocJar()
+    toolchain {
+        languageVersion = JavaLanguageVersion.of(25)
+        vendor = JvmVendorSpec.ADOPTIUM
+    }
 }
 
 compileJava {
diff --git a/build.gradle b/build.gradle
index 0b78d5d..b33ae81 100644
--- a/build.gradle
+++ b/build.gradle
@@ -3,8 +3,12 @@ import com.github.spotbugs.snom.Effort
 import org.eclipse.jgit.api.errors.RefAlreadyExistsException
 
 buildscript {
+    repositories {
+        google()
+        mavenCentral()
+    }
     if (!project.hasProperty("springBootVersion")) {
-        ext.springBootVersion = System.getenv("SPRING_BOOT_VERSION") ?: "3.4.6"
+        ext.springBootVersion = System.getenv("SPRING_BOOT_VERSION") ?: "4.0.6"
     }
     dependencies {
         classpath 'com.google.protobuf:protobuf-gradle-plugin:0.9.4'
@@ -15,11 +19,11 @@ buildscript {
 plugins {
     id 'idea'
     id 'org.springframework.boot' version "$springBootVersion" apply false
-    id "com.github.spotbugs" version "6.0.2"
+    id "com.github.spotbugs" version "6.5.5"
     id "at.zierler.yamlvalidator" version "1.5.0"
     id 'org.ajoberstar.grgit' version '5.2.1'
     id 'io.github.gradle-nexus.publish-plugin' version "1.3.0"
-    id 'com.github.johnrengelman.shadow' version '8.1.1' apply false
+    id 'com.gradleup.shadow' version '9.4.2' apply false
 }
 
 idea.project {
@@ -29,6 +33,7 @@ idea.project {
 }
 
 spotbugs {
+    toolVersion = '4.9.8'
     effort = Effort.valueOf('MAX')
     reportLevel = Confidence.valueOf('DEFAULT')
 }
diff --git a/build.libraries.gradle b/build.libraries.gradle
index 19c3ab7..2ed9cb8 100644
--- a/build.libraries.gradle
+++ b/build.libraries.gradle
@@ -1,6 +1,6 @@
 ext {
     protobufVersion = "3.24.0"
-    springBootVersion = "${System.getenv("SPRING_BOOT_VERSION") ?: '3.4.6'}"
+    springBootVersion = "${System.getenv("SPRING_BOOT_VERSION") ?: '4.0.6'}"
     libraries = [
             // version defined
             awaitility                      : 'org.awaitility:awaitility:4.2.0',
@@ -11,7 +11,7 @@ ext {
             jakartaValidationApi            : 'jakarta.validation:jakarta.validation-api:3.0.2',
             javaxValidationApi              : "javax.validation:validation-api:2.0.1.Final",
             kafkaStreams                    : 'org.apache.kafka:kafka-streams:3.2.3',
-            lz4Java                         : 'org.lz4:lz4-java:1.8.0',
+            lz4Java                         : 'at.yawk.lz4:lz4-java:1.10.1',
             protobufJava                    : "com.google.protobuf:protobuf-java:${protobufVersion}",
             semver4j                        : "com.vdurmont:semver4j:3.1.0",
             snappyJava                      : 'org.xerial.snappy:snappy-java:1.1.10.5',
@@ -32,7 +32,7 @@ ext {
             flywayCore                      : 'org.flywaydb:flyway-core',
             flywayMysql                     : 'org.flywaydb:flyway-mysql',
             flywayPostgresql                : 'org.flywaydb:flyway-database-postgresql',
-            jacksonDatabind                 : 'com.fasterxml.jackson.core:jackson-databind',
+            jacksonDatabind                 : 'tools.jackson.core:jackson-databind',
             kafkaClients                    : 'org.apache.kafka:kafka-clients',
             lombok                          : 'org.projectlombok:lombok',
             mariadbJavaClient               : 'org.mariadb.jdbc:mariadb-java-client',
@@ -40,6 +40,9 @@ ext {
             micrometerRegistryPrometheus    : 'io.micrometer:micrometer-registry-prometheus',
             postgresql                      : 'org.postgresql:postgresql',
             springBootConfigurationProcessor: 'org.springframework.boot:spring-boot-configuration-processor',
+            springBootFlyway                : 'org.springframework.boot:spring-boot-flyway',
+            springBootKafka                 : 'org.springframework.boot:spring-boot-kafka',
+            springBootValidation            : 'org.springframework.boot:spring-boot-validation',
             springBootStarter               : 'org.springframework.boot:spring-boot-starter',
             springBootStarterActuator       : 'org.springframework.boot:spring-boot-starter-actuator',
             springBootStarterJdbc           : 'org.springframework.boot:spring-boot-starter-jdbc',
diff --git a/demoapp/src/main/java/com/transferwise/kafka/tkms/demoapp/complextest/ComplexTestController.java b/demoapp/src/main/java/com/transferwise/kafka/tkms/demoapp/complextest/ComplexTestController.java
index be53b69..04810ec 100644
--- a/demoapp/src/main/java/com/transferwise/kafka/tkms/demoapp/complextest/ComplexTestController.java
+++ b/demoapp/src/main/java/com/transferwise/kafka/tkms/demoapp/complextest/ComplexTestController.java
@@ -1,6 +1,5 @@
 package com.transferwise.kafka.tkms.demoapp.complextest;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.transferwise.kafka.tkms.api.ITransactionalKafkaMessageSender;
 import com.transferwise.kafka.tkms.api.ITransactionalKafkaMessageSender.SendMessageResult;
 import com.transferwise.kafka.tkms.api.TkmsMessage;
@@ -10,6 +9,7 @@
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
+import tools.jackson.databind.ObjectMapper;
 
 @RestController
 public class ComplexTestController {
diff --git a/demoapp/src/main/java/com/transferwise/kafka/tkms/demoapp/complextest/ComplexTestMessagesListener.java b/demoapp/src/main/java/com/transferwise/kafka/tkms/demoapp/complextest/ComplexTestMessagesListener.java
index 81e4c9b..9dcf1ed 100644
--- a/demoapp/src/main/java/com/transferwise/kafka/tkms/demoapp/complextest/ComplexTestMessagesListener.java
+++ b/demoapp/src/main/java/com/transferwise/kafka/tkms/demoapp/complextest/ComplexTestMessagesListener.java
@@ -1,6 +1,5 @@
 package com.transferwise.kafka.tkms.demoapp.complextest;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.transferwise.common.baseutils.ExceptionUtils;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
@@ -13,6 +12,7 @@
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
+import tools.jackson.databind.ObjectMapper;
 
 @Slf4j
 @Component
diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar
index c1962a7..b1b8ef5 100644
Binary files a/gradle/wrapper/gradle-wrapper.jar and b/gradle/wrapper/gradle-wrapper.jar differ
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
index 3499ded..df6a6ad 100644
--- a/gradle/wrapper/gradle-wrapper.properties
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -1,6 +1,9 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.5-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-9.5.1-bin.zip
 networkTimeout=10000
+retries=0
+retryBackOffMs=500
+validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/gradlew b/gradlew
index aeb74cb..b9bb139 100755
--- a/gradlew
+++ b/gradlew
@@ -1,7 +1,7 @@
 #!/bin/sh
 
 #
-# Copyright © 2015-2021 the original authors.
+# Copyright © 2015 the original authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -15,6 +15,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+# SPDX-License-Identifier: Apache-2.0
+#
 
 ##############################################################################
 #
@@ -55,7 +57,7 @@
 #       Darwin, MinGW, and NonStop.
 #
 #   (3) This script is generated from the Groovy template
-#       https://github.com/gradle/gradle/blob/HEAD/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
+#       https://github.com/gradle/gradle/blob/3d91ce3b8caaf77ad09f381f43615b715b53f72c/platforms/jvm/plugins-application/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
 #       within the Gradle project.
 #
 #       You can find Gradle at https://github.com/gradle/gradle/.
@@ -83,7 +85,8 @@ done
 # This is normally unused
 # shellcheck disable=SC2034
 APP_BASE_NAME=${0##*/}
-APP_HOME=$( cd "${APP_HOME:-./}" && pwd -P ) || exit
+# Discard cd standard output in case $CDPATH is set (https://github.com/gradle/gradle/issues/25036)
+APP_HOME=$( cd -P "${APP_HOME:-./}" > /dev/null && printf '%s\n' "$PWD" ) || exit
 
 # Use the maximum available, or set MAX_FD != -1 to use that value.
 MAX_FD=maximum
@@ -111,7 +114,6 @@ case "$( uname )" in                #(
   NONSTOP* )        nonstop=true ;;
 esac
 
-CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
 
 
 # Determine the Java command to use to start the JVM.
@@ -130,10 +132,13 @@ location of your Java installation."
     fi
 else
     JAVACMD=java
-    which java >/dev/null 2>&1 || die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+    if ! command -v java >/dev/null 2>&1
+    then
+        die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
 
 Please set the JAVA_HOME variable in your environment to match the
 location of your Java installation."
+    fi
 fi
 
 # Increase the maximum file descriptors if we can.
@@ -141,7 +146,7 @@ if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then
     case $MAX_FD in #(
       max*)
         # In POSIX sh, ulimit -H is undefined. That's why the result is checked to see if it worked.
-        # shellcheck disable=SC3045
+        # shellcheck disable=SC2039,SC3045
         MAX_FD=$( ulimit -H -n ) ||
             warn "Could not query maximum file descriptor limit"
     esac
@@ -149,7 +154,7 @@ if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then
       '' | soft) :;; #(
       *)
         # In POSIX sh, ulimit -n is undefined. That's why the result is checked to see if it worked.
-        # shellcheck disable=SC3045
+        # shellcheck disable=SC2039,SC3045
         ulimit -n "$MAX_FD" ||
             warn "Could not set maximum file descriptor limit to $MAX_FD"
     esac
@@ -166,7 +171,6 @@ fi
 # For Cygwin or MSYS, switch paths to Windows format before running java
 if "$cygwin" || "$msys" ; then
     APP_HOME=$( cygpath --path --mixed "$APP_HOME" )
-    CLASSPATH=$( cygpath --path --mixed "$CLASSPATH" )
 
     JAVACMD=$( cygpath --unix "$JAVACMD" )
 
@@ -198,16 +202,15 @@ fi
 # Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
 DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
 
-# Collect all arguments for the java command;
-#   * $DEFAULT_JVM_OPTS, $JAVA_OPTS, and $GRADLE_OPTS can contain fragments of
-#     shell script including quotes and variable substitutions, so put them in
-#     double quotes to make sure that they get re-expanded; and
-#   * put everything else in single quotes, so that it's not re-expanded.
+# Collect all arguments for the java command:
+#   * DEFAULT_JVM_OPTS, JAVA_OPTS, and optsEnvironmentVar are not allowed to contain shell fragments,
+#     and any embedded shellness will be escaped.
+#   * For example: A user cannot expect ${Hostname} to be expanded, as it is an environment variable and will be
+#     treated as '${Hostname}' itself on the command line.
 
 set -- \
         "-Dorg.gradle.appname=$APP_BASE_NAME" \
-        -classpath "$CLASSPATH" \
-        org.gradle.wrapper.GradleWrapperMain \
+        -jar "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" \
         "$@"
 
 # Stop when "xargs" is not available.
diff --git a/gradlew.bat b/gradlew.bat
index 6689b85..aa5f10b 100644
--- a/gradlew.bat
+++ b/gradlew.bat
@@ -13,6 +13,8 @@
 @rem See the License for the specific language governing permissions and
 @rem limitations under the License.
 @rem
+@rem SPDX-License-Identifier: Apache-2.0
+@rem
 
 @if "%DEBUG%"=="" @echo off
 @rem ##########################################################################
@@ -21,8 +23,8 @@
 @rem
 @rem ##########################################################################
 
-@rem Set local scope for the variables with windows NT shell
-if "%OS%"=="Windows_NT" setlocal
+@rem Set local scope for the variables, and ensure extensions are enabled
+setlocal EnableExtensions
 
 set DIRNAME=%~dp0
 if "%DIRNAME%"=="" set DIRNAME=.
@@ -43,13 +45,13 @@ set JAVA_EXE=java.exe
 %JAVA_EXE% -version >NUL 2>&1
 if %ERRORLEVEL% equ 0 goto execute
 
-echo.
-echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
-echo.
-echo Please set the JAVA_HOME variable in your environment to match the
-echo location of your Java installation.
+echo. 1>&2
+echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. 1>&2
+echo. 1>&2
+echo Please set the JAVA_HOME variable in your environment to match the 1>&2
+echo location of your Java installation. 1>&2
 
-goto fail
+"%COMSPEC%" /c exit 1
 
 :findJavaFromJavaHome
 set JAVA_HOME=%JAVA_HOME:"=%
@@ -57,36 +59,24 @@ set JAVA_EXE=%JAVA_HOME%/bin/java.exe
 
 if exist "%JAVA_EXE%" goto execute
 
-echo.
-echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME%
-echo.
-echo Please set the JAVA_HOME variable in your environment to match the
-echo location of your Java installation.
+echo. 1>&2
+echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME% 1>&2
+echo. 1>&2
+echo Please set the JAVA_HOME variable in your environment to match the 1>&2
+echo location of your Java installation. 1>&2
 
-goto fail
+"%COMSPEC%" /c exit 1
 
 :execute
 @rem Setup the command line
 
-set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
 
 
 @rem Execute Gradle
-"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %*
-
-:end
-@rem End local scope for the variables with windows NT shell
-if %ERRORLEVEL% equ 0 goto mainEnd
-
-:fail
-rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of
-rem the _cmd.exe /c_ return code!
-set EXIT_CODE=%ERRORLEVEL%
-if %EXIT_CODE% equ 0 set EXIT_CODE=1
-if not ""=="%GRADLE_EXIT_CONSOLE%" exit %EXIT_CODE%
-exit /b %EXIT_CODE%
-
-:mainEnd
-if "%OS%"=="Windows_NT" endlocal
+@rem endlocal doesn't take effect until after the line is parsed and variables are expanded
+@rem which allows us to clear the local environment before executing the java command
+endlocal & "%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -jar "%APP_HOME%\gradle\wrapper\gradle-wrapper.jar" %* & call :exitWithErrorLevel
 
-:omega
+:exitWithErrorLevel
+@rem Use "%COMSPEC%" /c exit to allow operators to work properly in scripts
+"%COMSPEC%" /c exit %ERRORLEVEL%
diff --git a/mise.toml b/mise.toml
new file mode 100644
index 0000000..33b2c9b
--- /dev/null
+++ b/mise.toml
@@ -0,0 +1,2 @@
+[tools]
+java = "temurin-25"
diff --git a/settings.gradle b/settings.gradle
index 2d9d6f4..1eb0245 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -1,3 +1,6 @@
+plugins {
+    id 'org.gradle.toolchains.foojay-resolver-convention' version '1.0.0'
+}
 rootProject.name = "tw-tkms"
 include ':tw-tkms-starter'
 include ':tw-tkms-test-starter'
diff --git a/spotbugs-exclude.xml b/spotbugs-exclude.xml
index a54e4dc..1c6e3a0 100644
--- a/spotbugs-exclude.xml
+++ b/spotbugs-exclude.xml
@@ -5,11 +5,7 @@
   xsi:schemaLocation="https://github.com/spotbugs/filter/3.0.0 https://raw.githubusercontent.com/spotbugs/spotbugs/3.1.0/spotbugs/etc/findbugsfilter.xsd">
 
   <Match>
-    <Class name="com.transferwise.kafka.tkms.stored_message.StoredMessage$Header$Builder"/>
-  </Match>
-
-  <Match>
-    <Class name="com.transferwise.kafka.tkms.stored_message.StoredMessage$Message$Builder"/>
+    <Class name="~com\.transferwise\.kafka\.tkms\.stored_message\.StoredMessage\$.*"/>
   </Match>
 
   <Match>
diff --git a/tw-tkms-starter/build.gradle b/tw-tkms-starter/build.gradle
index 82f5228..54d9935 100644
--- a/tw-tkms-starter/build.gradle
+++ b/tw-tkms-starter/build.gradle
@@ -3,7 +3,7 @@ plugins {
     id 'idea'
     id "com.google.protobuf"
     id "docker-compose"
-    id 'com.github.johnrengelman.shadow'
+    id 'com.gradleup.shadow'
     id 'maven-publish'
     id 'signing'
 }
@@ -33,6 +33,8 @@ dependencies {
     implementation libraries.micrometerCore
     implementation libraries.protobufJava
     implementation libraries.springBootStarter
+    implementation libraries.springBootFlyway
+    implementation libraries.springBootValidation
     implementation libraries.springJdbc
     implementation libraries.twBaseUtils
     implementation libraries.twContext
@@ -67,6 +69,7 @@ dependencies {
     testImplementation libraries.springBootStarterTest
     testImplementation libraries.springBootStarterWeb
     testImplementation libraries.springKafka
+    testRuntimeOnly libraries.springBootKafka
 
     testRuntimeOnly libraries.mariadbJavaClient
     testRuntimeOnly libraries.postgresql
@@ -114,11 +117,7 @@ test {
  */
 shadowJar {
     dependencies {
-        dependencies {
-            exclude(dependency {
-                it.moduleName != 'protobuf-java'
-            })
-        }
+        include { it.moduleName == 'protobuf-java' }
     }
     manifest {
         attributes 'Implementation-Version': "$project.version"
diff --git a/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/api/ITransactionalKafkaMessageSender.java b/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/api/ITransactionalKafkaMessageSender.java
index aadc320..e86a758 100644
--- a/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/api/ITransactionalKafkaMessageSender.java
+++ b/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/api/ITransactionalKafkaMessageSender.java
@@ -14,6 +14,8 @@ public interface ITransactionalKafkaMessageSender {
    */
   SendMessageResult sendMessage(TkmsMessage message);
 
+  SendMessageResult sendMessage(SendMessageRequest request);
+
   @Data
   @Accessors(chain = true)
   class SendMessageResult {
@@ -34,7 +36,6 @@ class SendMessageResult {
     private TkmsShardPartition shardPartition;
   }
 
-  SendMessageResult sendMessage(SendMessageRequest request);
 
   @Data
   @Accessors(chain = true)
diff --git a/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/api/TkmsMessage.java b/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/api/TkmsMessage.java
index 14b4536..c7998d4 100644
--- a/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/api/TkmsMessage.java
+++ b/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/api/TkmsMessage.java
@@ -49,7 +49,7 @@ public class TkmsMessage {
   /**
    * Kafka message's content/payload.
    */
-  @SuppressFBWarnings({"EI_EXPOSE_REP", "EI_EXPOSE_REP2"})
+  @SuppressFBWarnings("EI_EXPOSE_REP")
   @NotNull
   private byte[] value;
 
diff --git a/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/api/helpers/TkmsMessageFactory.java b/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/api/helpers/TkmsMessageFactory.java
index 1698f67..316cb6e 100644
--- a/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/api/helpers/TkmsMessageFactory.java
+++ b/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/api/helpers/TkmsMessageFactory.java
@@ -1,11 +1,11 @@
 package com.transferwise.kafka.tkms.api.helpers;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.transferwise.common.baseutils.ExceptionUtils;
 import com.transferwise.kafka.tkms.api.TkmsMessage;
 import java.nio.charset.StandardCharsets;
 import lombok.NonNull;
 import org.springframework.beans.factory.annotation.Autowired;
+import tools.jackson.databind.ObjectMapper;
 
 public class TkmsMessageFactory implements ITkmsMessageFactory {
 
diff --git a/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/config/TkmsAutoConfiguration.java b/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/config/TkmsAutoConfiguration.java
index 987a80c..f0c0771 100644
--- a/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/config/TkmsAutoConfiguration.java
+++ b/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/config/TkmsAutoConfiguration.java
@@ -13,8 +13,8 @@
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.autoconfigure.flyway.FlywayAutoConfiguration;
-import org.springframework.boot.autoconfigure.validation.ValidationAutoConfiguration;
+import org.springframework.boot.flyway.autoconfigure.FlywayAutoConfiguration;
+import org.springframework.boot.validation.autoconfigure.ValidationAutoConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
diff --git a/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/dao/TkmsDao.java b/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/dao/TkmsDao.java
index 121f31a..ddab063 100644
--- a/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/dao/TkmsDao.java
+++ b/tw-tkms-starter/src/main/java/com/transferwise/kafka/tkms/dao/TkmsDao.java
@@ -9,7 +9,6 @@
 import com.transferwise.kafka.tkms.api.TkmsShardPartition;
 import com.transferwise.kafka.tkms.config.TkmsProperties;
 import com.transferwise.kafka.tkms.metrics.ITkmsMetricsTemplate;
-import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.io.InputStream;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
@@ -195,7 +194,6 @@ protected InputStream serializeMessage(TkmsShardPartition shardPartition, TkmsMe
     return ExceptionUtils.doUnchecked(() -> messageSerializer.serialize(shardPartition, message));
   }
 
-  @SuppressFBWarnings("NP_NULL_ON_SOME_PATH_FROM_RETURN_VALUE")
   protected long keyToLong(KeyHolder keyHolder) {
     return keyHolder.getKey().longValue();
   }
diff --git a/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/EndToEndIntTest.java b/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/EndToEndIntTest.java
index 51dc59d..3602567 100644
--- a/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/EndToEndIntTest.java
+++ b/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/EndToEndIntTest.java
@@ -4,7 +4,6 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.awaitility.Awaitility.await;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.transferwise.common.baseutils.ExceptionUtils;
 import com.transferwise.common.baseutils.UuidUtils;
 import com.transferwise.common.baseutils.transactionsmanagement.ITransactionsHelper;
@@ -49,6 +48,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.springframework.beans.factory.annotation.Autowired;
+import tools.jackson.databind.ObjectMapper;
 
 @BaseTestEnvironment
 @Slf4j
diff --git a/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/test/FlywayConfiguration.java b/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/test/FlywayConfiguration.java
index b0deb18..0373904 100644
--- a/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/test/FlywayConfiguration.java
+++ b/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/test/FlywayConfiguration.java
@@ -1,6 +1,6 @@
 package com.transferwise.kafka.tkms.test;
 
-import org.springframework.boot.autoconfigure.flyway.FlywayMigrationStrategy;
+import org.springframework.boot.flyway.autoconfigure.FlywayMigrationStrategy;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Primary;
diff --git a/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/test/KafkaConfiguration.java b/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/test/KafkaConfiguration.java
index 8c93059..4204a4f 100644
--- a/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/test/KafkaConfiguration.java
+++ b/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/test/KafkaConfiguration.java
@@ -2,7 +2,6 @@
 
 import com.transferwise.common.baseutils.ExceptionUtils;
 import com.transferwise.kafka.tkms.test.TestProperties.KafkaServer;
-import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -29,7 +28,6 @@ public class KafkaConfiguration implements InitializingBean {
   @Autowired
   private TestProperties tkmsProperties;
 
-  @SuppressFBWarnings("RCN_REDUNDANT_NULLCHECK_WOULD_HAVE_BEEN_A_NPE")
   @Override
   public void afterPropertiesSet() {
     for (var kafkaServer : tkmsProperties.getKafkaServers()) {
@@ -73,6 +71,7 @@ protected void createTopic(final AdminClient adminClient, final String topicName
         final CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singleton(newTopic));
         createTopicsResult.values().get(topicName).get();
         log.info("Created Kafka topic '" + topicName + "' in " + i + ".");
+        awaitTopicMetadataPropagated(adminClient, topicName, partitions);
         return;
       } catch (InterruptedException | ExecutionException e) {
         if (!(e.getCause() instanceof TopicExistsException)) {
@@ -83,4 +82,29 @@ protected void createTopic(final AdminClient adminClient, final String topicName
     }
   }
 
+  /**
+   * `createTopics` acknowledges as soon as the topic is created in the controller, but partition leadership is not necessarily propagated to all
+   * brokers yet. TKMS validates topics via the producer's metadata at startup, which fails if it queries before propagation completes. Wait until the
+   * topic is fully describable with a leader for every partition, so topic validation does not flake under load.
+   */
+  protected void awaitTopicMetadataPropagated(final AdminClient adminClient, final String topicName, final int partitions) {
+    for (int i = 0; i < 100; i++) {
+      try {
+        var description = adminClient.describeTopics(Collections.singleton(topicName)).allTopicNames().get().get(topicName);
+        boolean ready = description != null
+            && description.partitions().size() == partitions
+            && description.partitions().stream().allMatch(p -> p.leader() != null && p.leader().id() >= 0);
+        if (ready) {
+          return;
+        }
+      } catch (InterruptedException | ExecutionException e) {
+        if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) {
+          throw new RuntimeException(e.getMessage(), e);
+        }
+      }
+      ExceptionUtils.doUnchecked(() -> Thread.sleep(50));
+    }
+    throw new IllegalStateException("Topic '" + topicName + "' metadata did not propagate in time.");
+  }
+
 }
diff --git a/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/test/TestApplication.java b/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/test/TestApplication.java
index ac71093..72234c3 100644
--- a/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/test/TestApplication.java
+++ b/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/test/TestApplication.java
@@ -4,7 +4,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.boot.autoconfigure.flyway.FlywayMigrationStrategy;
+import org.springframework.boot.flyway.autoconfigure.FlywayMigrationStrategy;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
 
diff --git a/tw-tkms-starter/src/test/resources/docker-compose.yml b/tw-tkms-starter/src/test/resources/docker-compose.yml
index 3b7a4eb..c84b6d8 100644
--- a/tw-tkms-starter/src/test/resources/docker-compose.yml
+++ b/tw-tkms-starter/src/test/resources/docker-compose.yml
@@ -9,55 +9,45 @@ services:
       ALLOW_ANONYMOUS_LOGIN: "yes"
       JVMFLAGS: -server -Xms25m -Xmx512m -XX:+HeapDumpOnOutOfMemoryError -XX:GCHeapFreeLimit=5 -XX:GCTimeLimit=90  -XX:SoftRefLRUPolicyMSPerMB=5 -XX:MinHeapFreeRatio=10 -XX:MaxHeapFreeRatio=20 -XX:+ExplicitGCInvokesConcurrent
   kafka1:
-    image: wurstmeister/kafka:2.12-2.4.1
+    image: bitnamilegacy/kafka:3.6.0
     ports:
-      - "9092"
+      - "9092:9092"
     container_name: "tw_tkms_kafka_1"
     environment:
-      PORT_COMMAND: "docker port $$(docker ps -q -f name=tw_tkms_kafka_1) 9092/tcp | cut -d: -f2"
-      KAFKA_BROKER_ID: 1
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181/kafka1
-      KAFKA_ADVERTISED_LISTENERS: "EXTERNAL://localhost:_{PORT_COMMAND},INTERNAL://kafka1:9093"
-      KAFKA_LISTENERS: EXTERNAL://:9092,INTERNAL://:9093
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
-      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-      KAFKA_MESSAGE_MAX_BYTES: 10485760
-      KAFKA_ZOOKEEPER_SESSION_TIMEOUT_MS: 20000
-      KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE: "true"
-      KAFKA_LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS: 5
-      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
-      LOG4J_LOGGER_ORG: WARN,STDOUT
-      LOG4J_LOGGER_ORG_APACHE_KAFKA: WARN,STDOUT
-      LOG4J_LOGGER_KAFKA: WARN,STDOUT
-      KAFKA_JVM_PERFORMANCE_OPTS: -server -Xms25m -Xmx512m -XX:+HeapDumpOnOutOfMemoryError -XX:GCHeapFreeLimit=5 -XX:GCTimeLimit=90  -XX:SoftRefLRUPolicyMSPerMB=5 -XX:MinHeapFreeRatio=10 -XX:MaxHeapFreeRatio=20 -XX:+ExplicitGCInvokesConcurrent
-    volumes:
-      - /var/run/docker.sock:/var/run/docker.sock
+      ALLOW_PLAINTEXT_LISTENER: "yes"
+      KAFKA_CFG_BROKER_ID: 1
+      KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181/kafka1
+      KAFKA_CFG_LISTENERS: EXTERNAL://:9092,INTERNAL://:9093
+      KAFKA_CFG_ADVERTISED_LISTENERS: "EXTERNAL://localhost:9092,INTERNAL://kafka1:9093"
+      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
+      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_CFG_MESSAGE_MAX_BYTES: 10485760
+      KAFKA_CFG_ZOOKEEPER_SESSION_TIMEOUT_MS: 20000
+      KAFKA_CFG_UNCLEAN_LEADER_ELECTION_ENABLE: "true"
+      KAFKA_CFG_LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS: 5
+      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "false"
+      KAFKA_HEAP_OPTS: -Xms256m -Xmx768m
   kafka2:
-    image: wurstmeister/kafka:2.12-2.4.1
+    image: bitnamilegacy/kafka:3.6.0
     ports:
-      - "9092"
+      - "9093:9092"
     container_name: "tw_tkms_kafka_2"
     environment:
-      PORT_COMMAND: "docker port $$(docker ps -q -f name=tw_tkms_kafka_2) 9092/tcp | cut -d: -f2"
-      KAFKA_BROKER_ID: 2
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181/kafka2
-      KAFKA_ADVERTISED_LISTENERS: "EXTERNAL://localhost:_{PORT_COMMAND},INTERNAL://kafka2:9093"
-      KAFKA_LISTENERS: EXTERNAL://:9092,INTERNAL://:9093
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
-      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-      KAFKA_MESSAGE_MAX_BYTES: 10485760
-      KAFKA_ZOOKEEPER_SESSION_TIMEOUT_MS: 20000
-      KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE: "true"
-      KAFKA_LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS: 5
-      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
-      LOG4J_LOGGER_ORG: WARN,STDOUT
-      LOG4J_LOGGER_ORG_APACHE_KAFKA: WARN,STDOUT
-      LOG4J_LOGGER_KAFKA: WARN,STDOUT
-      KAFKA_JVM_PERFORMANCE_OPTS: -server -Xms25m -Xmx512m -XX:+HeapDumpOnOutOfMemoryError -XX:GCHeapFreeLimit=5 -XX:GCTimeLimit=90  -XX:SoftRefLRUPolicyMSPerMB=5 -XX:MinHeapFreeRatio=10 -XX:MaxHeapFreeRatio=20 -XX:+ExplicitGCInvokesConcurrent
-    volumes:
-      - /var/run/docker.sock:/var/run/docker.sock
+      ALLOW_PLAINTEXT_LISTENER: "yes"
+      KAFKA_CFG_BROKER_ID: 2
+      KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181/kafka2
+      KAFKA_CFG_LISTENERS: EXTERNAL://:9092,INTERNAL://:9093
+      KAFKA_CFG_ADVERTISED_LISTENERS: "EXTERNAL://localhost:9093,INTERNAL://kafka2:9093"
+      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
+      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_CFG_MESSAGE_MAX_BYTES: 10485760
+      KAFKA_CFG_ZOOKEEPER_SESSION_TIMEOUT_MS: 20000
+      KAFKA_CFG_UNCLEAN_LEADER_ELECTION_ENABLE: "true"
+      KAFKA_CFG_LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS: 5
+      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "false"
+      KAFKA_HEAP_OPTS: -Xms256m -Xmx768m
   mariadb:
     image: mariadb:10.6
     ports:
diff --git a/tw-tkms-test-starter/src/main/java/com/transferwise/kafka/tkms/test/TkmsRegisteredMessagesCollector.java b/tw-tkms-test-starter/src/main/java/com/transferwise/kafka/tkms/test/TkmsRegisteredMessagesCollector.java
index 601101c..cea7a9b 100644
--- a/tw-tkms-test-starter/src/main/java/com/transferwise/kafka/tkms/test/TkmsRegisteredMessagesCollector.java
+++ b/tw-tkms-test-starter/src/main/java/com/transferwise/kafka/tkms/test/TkmsRegisteredMessagesCollector.java
@@ -1,6 +1,5 @@
 package com.transferwise.kafka.tkms.test;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.transferwise.common.baseutils.ExceptionUtils;
 import com.transferwise.kafka.tkms.api.ITkmsEventsListener;
 import com.transferwise.kafka.tkms.api.TkmsShardPartition;
@@ -14,6 +13,7 @@
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.tuple.Pair;
 import org.springframework.beans.factory.annotation.Autowired;
+import tools.jackson.databind.ObjectMapper;
 
 public class TkmsRegisteredMessagesCollector implements ITkmsRegisteredMessagesCollector, ITkmsEventsListener {
 
@@ -26,7 +26,7 @@ public class TkmsRegisteredMessagesCollector implements ITkmsRegisteredMessagesC
 
   private AtomicInteger messagesCount = new AtomicInteger();
 
-  private boolean enabled = true;
+  private volatile boolean enabled = true;
 
   @Override
   public void messageRegistered(MessageRegisteredEvent event) {
diff --git a/tw-tkms-test-starter/src/main/java/com/transferwise/kafka/tkms/test/TkmsSentMessagesCollector.java b/tw-tkms-test-starter/src/main/java/com/transferwise/kafka/tkms/test/TkmsSentMessagesCollector.java
index 6649a82..19972be 100644
--- a/tw-tkms-test-starter/src/main/java/com/transferwise/kafka/tkms/test/TkmsSentMessagesCollector.java
+++ b/tw-tkms-test-starter/src/main/java/com/transferwise/kafka/tkms/test/TkmsSentMessagesCollector.java
@@ -1,6 +1,5 @@
 package com.transferwise.kafka.tkms.test;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.transferwise.common.baseutils.ExceptionUtils;
 import com.transferwise.kafka.tkms.api.ITkmsEventsListener;
 import com.transferwise.kafka.tkms.api.TkmsShardPartition;
@@ -14,6 +13,7 @@
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.tuple.Pair;
 import org.springframework.beans.factory.annotation.Autowired;
+import tools.jackson.databind.ObjectMapper;
 
 public class TkmsSentMessagesCollector implements ITkmsSentMessagesCollector, ITkmsEventsListener {