GH-666: Single global embedded Kafka with JUnit 5 (#2308)

* GH-666: Single global embedded Kafka with JUnit 5 Fixes https://github.com/spring-projects/spring-kafka/issues/666 * Introduce a `GlobalEmbeddedKafkaTestExecutionListener` which is registered by the service loader from JUnit Platform * Make its activation conditional based on the `spring.kafka.global.embedded.enabled` system property * Expose some other configuration properties for the target global `EmbeddedKafkaBroker` * Verify its functionality via manual `Launcher.execute()` * Add more `@DirtiesContext` to some tests in the `spring-kafka-test` which don't close their embedded brokers on the exit * * Print local launcher summary into SOUT for tracing the nested problems * * Fix prefix for `file:` resource in the test * * Make `junit-platform-launcher` as provided dep - this is part of JUnit platform the test plan is going to be run * Add missed new line in the end of `junit-platform.properties` * Rethrow an exception from the local launcher in the `GlobalEmbeddedKafkaTestExecutionListenerTests` to fail if its suite test classes have failed. * * Fix JavaDocs for `GlobalEmbeddedKafkaTestExecutionListener` constants * Move system properties clean up in the `GlobalEmbeddedKafkaTestExecutionListenerTests` into `@AfterAll` * Add docs for this new feature * Add `sample-05` to demonstrate global embedded Kafka in action via Maven configuration * * Mention `sample-05` in the common `README.adoc` for samples * * Fix language in the `testing.adoc` * Add new lines into new files * Add `spring.kafka.consumer.auto-offset-reset=earliest` to make a consumer to start from the beginning of the partition: fixes a race condition when publishing happens before consumer is started * Make `mvnw` as an executable * Add `spring.embedded.kafka.brokers.property` explanation into docs * * Fix language in docs Co-authored-by: Gary Russell <grussell@vmware.com> * * Make `Sample05Application2Tests` passing * Remove `deliberately failing` sentence from the JavaDoc an README * Clean up unused imports Co-authored-by: Gary Russell <grussell@vmware.com>
2022-06-22 17:14:14 -04:00
parent 596f7d940c
commit 0f69553736
24 changed files with 1212 additions and 5 deletions
--- a/build.gradle
+++ b/build.gradle
@@ -373,6 +373,7 @@ project ('spring-kafka-test') {
 		api "org.apache.kafka:kafka_$scalaVersion:$kafkaVersion"
 		api "org.apache.kafka:kafka_$scalaVersion:$kafkaVersion:test"
 		api 'org.junit.jupiter:junit-jupiter-api'
+		providedApi 'org.junit.platform:junit-platform-launcher'
 		optionalApi "org.hamcrest:hamcrest-core:$hamcrestVersion"
 		optionalApi "org.mockito:mockito-core:$mockitoVersion"
 		optionalApi ("junit:junit:$junit4Version") {
--- a/samples/README.adoc
+++ b/samples/README.adoc
@@ -4,3 +4,4 @@
 * sample-02 - multi-method listener
 * sample-03 - transactions
 * sample-04 - topic based (non-blocking) retry
+* sample-05 - global embedded Kafka testing
--- a/samples/sample-05/.gitignore
+++ b/samples/sample-05/.gitignore
@@ -0,0 +1,33 @@
+HELP.md
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/
--- a/samples/sample-05/.mvn/wrapper/maven-wrapper.jar
+++ b/samples/sample-05/.mvn/wrapper/maven-wrapper.jar
--- a/samples/sample-05/.mvn/wrapper/maven-wrapper.properties
+++ b/samples/sample-05/.mvn/wrapper/maven-wrapper.properties
@@ -0,0 +1,2 @@
+distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.8.5/apache-maven-3.8.5-bin.zip
+wrapperUrl=https://repo.maven.apache.org/maven2/org/apache/maven/wrapper/maven-wrapper/3.1.0/maven-wrapper-3.1.0.jar
--- a/samples/sample-05/README.adoc
+++ b/samples/sample-05/README.adoc
@@ -0,0 +1,23 @@
+== Sample 5
+
+This sample demonstrates a global embedded Kafka broker in action.
+
+This kind of broker is enabled via `spring.kafka.global.embedded.enabled = true` JUnit Platform property configured in the Maven Surefire plugin.
+Additional properties for the underlying `GlobalEmbeddedKafkaTestExecutionListener`, and therefore embedded Kafka broker(s), are provided via standard `junit-platform.properties` file or extra configuration parameters for the Maven Surefire plugin.
+
+Run only `./mvnw test` command for this sample.
+The Maven will report to the output similar two log messages:
+```
+11:03:44.383 [main] INFO  o.s.k.t.j.GlobalEmbeddedKafkaTestExecutionListener - Started global Embedded Kafka on: 127.0.0.1:53671
+...
+11:03:48.439 [main] INFO  o.s.k.t.j.GlobalEmbeddedKafkaTestExecutionListener - Stopped global Embedded Kafka.
+```
+One in the beginning of the test plan - and another in the end before reporting failures.
+
+The application by itself is going to fail because it has a configuration like `spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}`.
+This property makes it available only when an `EmbeddedKafkaBroker` is started, which is done by the `GlobalEmbeddedKafkaTestExecutionListener` and only when `spring.kafka.global.embedded.enabled = true`.
+Running unit tests from IDE will lead to similar failure -  the `GlobalEmbeddedKafkaTestExecutionListener` is not enabled by default and there is no an `@EmbeddedKafka` present in these tests.
+
+The `Sample05Application2Tests` demonstrates that global embedded Kafka broker config for {@code auto.create.topics.enable=false} is in an effect.
+See `kafka-broker.properties` added via Maven Surefire plugin.
+See `/test/resources/kafka-broker.properties` and Maven Surefire plugin configuration in the pom.xml.
--- a/samples/sample-05/mvnw
+++ b/samples/sample-05/mvnw
@@ -0,0 +1,316 @@
+#!/bin/sh
+# ----------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# ----------------------------------------------------------------------------
+
+# ----------------------------------------------------------------------------
+# Maven Start Up Batch script
+#
+# Required ENV vars:
+# ------------------
+#   JAVA_HOME - location of a JDK home dir
+#
+# Optional ENV vars
+# -----------------
+#   M2_HOME - location of maven2's installed home dir
+#   MAVEN_OPTS - parameters passed to the Java VM when running Maven
+#     e.g. to debug Maven itself, use
+#       set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
+#   MAVEN_SKIP_RC - flag to disable loading of mavenrc files
+# ----------------------------------------------------------------------------
+
+if [ -z "$MAVEN_SKIP_RC" ] ; then
+
+  if [ -f /usr/local/etc/mavenrc ] ; then
+    . /usr/local/etc/mavenrc
+  fi
+
+  if [ -f /etc/mavenrc ] ; then
+    . /etc/mavenrc
+  fi
+
+  if [ -f "$HOME/.mavenrc" ] ; then
+    . "$HOME/.mavenrc"
+  fi
+
+fi
+
+# OS specific support.  $var _must_ be set to either true or false.
+cygwin=false;
+darwin=false;
+mingw=false
+case "`uname`" in
+  CYGWIN*) cygwin=true ;;
+  MINGW*) mingw=true;;
+  Darwin*) darwin=true
+    # Use /usr/libexec/java_home if available, otherwise fall back to /Library/Java/Home
+    # See https://developer.apple.com/library/mac/qa/qa1170/_index.html
+    if [ -z "$JAVA_HOME" ]; then
+      if [ -x "/usr/libexec/java_home" ]; then
+        export JAVA_HOME="`/usr/libexec/java_home`"
+      else
+        export JAVA_HOME="/Library/Java/Home"
+      fi
+    fi
+    ;;
+esac
+
+if [ -z "$JAVA_HOME" ] ; then
+  if [ -r /etc/gentoo-release ] ; then
+    JAVA_HOME=`java-config --jre-home`
+  fi
+fi
+
+if [ -z "$M2_HOME" ] ; then
+  ## resolve links - $0 may be a link to maven's home
+  PRG="$0"
+
+  # need this for relative symlinks
+  while [ -h "$PRG" ] ; do
+    ls=`ls -ld "$PRG"`
+    link=`expr "$ls" : '.*-> \(.*\)$'`
+    if expr "$link" : '/.*' > /dev/null; then
+      PRG="$link"
+    else
+      PRG="`dirname "$PRG"`/$link"
+    fi
+  done
+
+  saveddir=`pwd`
+
+  M2_HOME=`dirname "$PRG"`/..
+
+  # make it fully qualified
+  M2_HOME=`cd "$M2_HOME" && pwd`
+
+  cd "$saveddir"
+  # echo Using m2 at $M2_HOME
+fi
+
+# For Cygwin, ensure paths are in UNIX format before anything is touched
+if $cygwin ; then
+  [ -n "$M2_HOME" ] &&
+    M2_HOME=`cygpath --unix "$M2_HOME"`
+  [ -n "$JAVA_HOME" ] &&
+    JAVA_HOME=`cygpath --unix "$JAVA_HOME"`
+  [ -n "$CLASSPATH" ] &&
+    CLASSPATH=`cygpath --path --unix "$CLASSPATH"`
+fi
+
+# For Mingw, ensure paths are in UNIX format before anything is touched
+if $mingw ; then
+  [ -n "$M2_HOME" ] &&
+    M2_HOME="`(cd "$M2_HOME"; pwd)`"
+  [ -n "$JAVA_HOME" ] &&
+    JAVA_HOME="`(cd "$JAVA_HOME"; pwd)`"
+fi
+
+if [ -z "$JAVA_HOME" ]; then
+  javaExecutable="`which javac`"
+  if [ -n "$javaExecutable" ] && ! [ "`expr \"$javaExecutable\" : '\([^ ]*\)'`" = "no" ]; then
+    # readlink(1) is not available as standard on Solaris 10.
+    readLink=`which readlink`
+    if [ ! `expr "$readLink" : '\([^ ]*\)'` = "no" ]; then
+      if $darwin ; then
+        javaHome="`dirname \"$javaExecutable\"`"
+        javaExecutable="`cd \"$javaHome\" && pwd -P`/javac"
+      else
+        javaExecutable="`readlink -f \"$javaExecutable\"`"
+      fi
+      javaHome="`dirname \"$javaExecutable\"`"
+      javaHome=`expr "$javaHome" : '\(.*\)/bin'`
+      JAVA_HOME="$javaHome"
+      export JAVA_HOME
+    fi
+  fi
+fi
+
+if [ -z "$JAVACMD" ] ; then
+  if [ -n "$JAVA_HOME"  ] ; then
+    if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
+      # IBM's JDK on AIX uses strange locations for the executables
+      JAVACMD="$JAVA_HOME/jre/sh/java"
+    else
+      JAVACMD="$JAVA_HOME/bin/java"
+    fi
+  else
+    JAVACMD="`\\unset -f command; \\command -v java`"
+  fi
+fi
+
+if [ ! -x "$JAVACMD" ] ; then
+  echo "Error: JAVA_HOME is not defined correctly." >&2
+  echo "  We cannot execute $JAVACMD" >&2
+  exit 1
+fi
+
+if [ -z "$JAVA_HOME" ] ; then
+  echo "Warning: JAVA_HOME environment variable is not set."
+fi
+
+CLASSWORLDS_LAUNCHER=org.codehaus.plexus.classworlds.launcher.Launcher
+
+# traverses directory structure from process work directory to filesystem root
+# first directory with .mvn subdirectory is considered project base directory
+find_maven_basedir() {
+
+  if [ -z "$1" ]
+  then
+    echo "Path not specified to find_maven_basedir"
+    return 1
+  fi
+
+  basedir="$1"
+  wdir="$1"
+  while [ "$wdir" != '/' ] ; do
+    if [ -d "$wdir"/.mvn ] ; then
+      basedir=$wdir
+      break
+    fi
+    # workaround for JBEAP-8937 (on Solaris 10/Sparc)
+    if [ -d "${wdir}" ]; then
+      wdir=`cd "$wdir/.."; pwd`
+    fi
+    # end of workaround
+  done
+  echo "${basedir}"
+}
+
+# concatenates all lines of a file
+concat_lines() {
+  if [ -f "$1" ]; then
+    echo "$(tr -s '\n' ' ' < "$1")"
+  fi
+}
+
+BASE_DIR=`find_maven_basedir "$(pwd)"`
+if [ -z "$BASE_DIR" ]; then
+  exit 1;
+fi
+
+##########################################################################################
+# Extension to allow automatically downloading the maven-wrapper.jar from Maven-central
+# This allows using the maven wrapper in projects that prohibit checking in binary data.
+##########################################################################################
+if [ -r "$BASE_DIR/.mvn/wrapper/maven-wrapper.jar" ]; then
+    if [ "$MVNW_VERBOSE" = true ]; then
+      echo "Found .mvn/wrapper/maven-wrapper.jar"
+    fi
+else
+    if [ "$MVNW_VERBOSE" = true ]; then
+      echo "Couldn't find .mvn/wrapper/maven-wrapper.jar, downloading it ..."
+    fi
+    if [ -n "$MVNW_REPOURL" ]; then
+      jarUrl="$MVNW_REPOURL/org/apache/maven/wrapper/maven-wrapper/3.1.0/maven-wrapper-3.1.0.jar"
+    else
+      jarUrl="https://repo.maven.apache.org/maven2/org/apache/maven/wrapper/maven-wrapper/3.1.0/maven-wrapper-3.1.0.jar"
+    fi
+    while IFS="=" read key value; do
+      case "$key" in (wrapperUrl) jarUrl="$value"; break ;;
+      esac
+    done < "$BASE_DIR/.mvn/wrapper/maven-wrapper.properties"
+    if [ "$MVNW_VERBOSE" = true ]; then
+      echo "Downloading from: $jarUrl"
+    fi
+    wrapperJarPath="$BASE_DIR/.mvn/wrapper/maven-wrapper.jar"
+    if $cygwin; then
+      wrapperJarPath=`cygpath --path --windows "$wrapperJarPath"`
+    fi
+
+    if command -v wget > /dev/null; then
+        if [ "$MVNW_VERBOSE" = true ]; then
+          echo "Found wget ... using wget"
+        fi
+        if [ -z "$MVNW_USERNAME" ] || [ -z "$MVNW_PASSWORD" ]; then
+            wget "$jarUrl" -O "$wrapperJarPath" || rm -f "$wrapperJarPath"
+        else
+            wget --http-user=$MVNW_USERNAME --http-password=$MVNW_PASSWORD "$jarUrl" -O "$wrapperJarPath" || rm -f "$wrapperJarPath"
+        fi
+    elif command -v curl > /dev/null; then
+        if [ "$MVNW_VERBOSE" = true ]; then
+          echo "Found curl ... using curl"
+        fi
+        if [ -z "$MVNW_USERNAME" ] || [ -z "$MVNW_PASSWORD" ]; then
+            curl -o "$wrapperJarPath" "$jarUrl" -f
+        else
+            curl --user $MVNW_USERNAME:$MVNW_PASSWORD -o "$wrapperJarPath" "$jarUrl" -f
+        fi
+
+    else
+        if [ "$MVNW_VERBOSE" = true ]; then
+          echo "Falling back to using Java to download"
+        fi
+        javaClass="$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.java"
+        # For Cygwin, switch paths to Windows format before running javac
+        if $cygwin; then
+          javaClass=`cygpath --path --windows "$javaClass"`
+        fi
+        if [ -e "$javaClass" ]; then
+            if [ ! -e "$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.class" ]; then
+                if [ "$MVNW_VERBOSE" = true ]; then
+                  echo " - Compiling MavenWrapperDownloader.java ..."
+                fi
+                # Compiling the Java class
+                ("$JAVA_HOME/bin/javac" "$javaClass")
+            fi
+            if [ -e "$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.class" ]; then
+                # Running the downloader
+                if [ "$MVNW_VERBOSE" = true ]; then
+                  echo " - Running MavenWrapperDownloader.java ..."
+                fi
+                ("$JAVA_HOME/bin/java" -cp .mvn/wrapper MavenWrapperDownloader "$MAVEN_PROJECTBASEDIR")
+            fi
+        fi
+    fi
+fi
+##########################################################################################
+# End of extension
+##########################################################################################
+
+export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-"$BASE_DIR"}
+if [ "$MVNW_VERBOSE" = true ]; then
+  echo $MAVEN_PROJECTBASEDIR
+fi
+MAVEN_OPTS="$(concat_lines "$MAVEN_PROJECTBASEDIR/.mvn/jvm.config") $MAVEN_OPTS"
+
+# For Cygwin, switch paths to Windows format before running java
+if $cygwin; then
+  [ -n "$M2_HOME" ] &&
+    M2_HOME=`cygpath --path --windows "$M2_HOME"`
+  [ -n "$JAVA_HOME" ] &&
+    JAVA_HOME=`cygpath --path --windows "$JAVA_HOME"`
+  [ -n "$CLASSPATH" ] &&
+    CLASSPATH=`cygpath --path --windows "$CLASSPATH"`
+  [ -n "$MAVEN_PROJECTBASEDIR" ] &&
+    MAVEN_PROJECTBASEDIR=`cygpath --path --windows "$MAVEN_PROJECTBASEDIR"`
+fi
+
+# Provide a "standardized" way to retrieve the CLI args that will
+# work with both Windows and non-Windows executions.
+MAVEN_CMD_LINE_ARGS="$MAVEN_CONFIG $@"
+export MAVEN_CMD_LINE_ARGS
+
+WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
+
+exec "$JAVACMD" \
+  $MAVEN_OPTS \
+  $MAVEN_DEBUG_OPTS \
+  -classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \
+  "-Dmaven.home=${M2_HOME}" \
+  "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \
+  ${WRAPPER_LAUNCHER} $MAVEN_CONFIG "$@"
--- a/samples/sample-05/mvnw.cmd
+++ b/samples/sample-05/mvnw.cmd
@@ -0,0 +1,188 @@
+@REM ----------------------------------------------------------------------------
+@REM Licensed to the Apache Software Foundation (ASF) under one
+@REM or more contributor license agreements.  See the NOTICE file
+@REM distributed with this work for additional information
+@REM regarding copyright ownership.  The ASF licenses this file
+@REM to you under the Apache License, Version 2.0 (the
+@REM "License"); you may not use this file except in compliance
+@REM with the License.  You may obtain a copy of the License at
+@REM
+@REM    https://www.apache.org/licenses/LICENSE-2.0
+@REM
+@REM Unless required by applicable law or agreed to in writing,
+@REM software distributed under the License is distributed on an
+@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+@REM KIND, either express or implied.  See the License for the
+@REM specific language governing permissions and limitations
+@REM under the License.
+@REM ----------------------------------------------------------------------------
+
+@REM ----------------------------------------------------------------------------
+@REM Maven Start Up Batch script
+@REM
+@REM Required ENV vars:
+@REM JAVA_HOME - location of a JDK home dir
+@REM
+@REM Optional ENV vars
+@REM M2_HOME - location of maven2's installed home dir
+@REM MAVEN_BATCH_ECHO - set to 'on' to enable the echoing of the batch commands
+@REM MAVEN_BATCH_PAUSE - set to 'on' to wait for a keystroke before ending
+@REM MAVEN_OPTS - parameters passed to the Java VM when running Maven
+@REM     e.g. to debug Maven itself, use
+@REM set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
+@REM MAVEN_SKIP_RC - flag to disable loading of mavenrc files
+@REM ----------------------------------------------------------------------------
+
+@REM Begin all REM lines with '@' in case MAVEN_BATCH_ECHO is 'on'
+@echo off
+@REM set title of command window
+title %0
+@REM enable echoing by setting MAVEN_BATCH_ECHO to 'on'
+@if "%MAVEN_BATCH_ECHO%" == "on"  echo %MAVEN_BATCH_ECHO%
+
+@REM set %HOME% to equivalent of $HOME
+if "%HOME%" == "" (set "HOME=%HOMEDRIVE%%HOMEPATH%")
+
+@REM Execute a user defined script before this one
+if not "%MAVEN_SKIP_RC%" == "" goto skipRcPre
+@REM check for pre script, once with legacy .bat ending and once with .cmd ending
+if exist "%USERPROFILE%\mavenrc_pre.bat" call "%USERPROFILE%\mavenrc_pre.bat" %*
+if exist "%USERPROFILE%\mavenrc_pre.cmd" call "%USERPROFILE%\mavenrc_pre.cmd" %*
+:skipRcPre
+
+@setlocal
+
+set ERROR_CODE=0
+
+@REM To isolate internal variables from possible post scripts, we use another setlocal
+@setlocal
+
+@REM ==== START VALIDATION ====
+if not "%JAVA_HOME%" == "" goto OkJHome
+
+echo.
+echo Error: JAVA_HOME not found in your environment. >&2
+echo Please set the JAVA_HOME variable in your environment to match the >&2
+echo location of your Java installation. >&2
+echo.
+goto error
+
+:OkJHome
+if exist "%JAVA_HOME%\bin\java.exe" goto init
+
+echo.
+echo Error: JAVA_HOME is set to an invalid directory. >&2
+echo JAVA_HOME = "%JAVA_HOME%" >&2
+echo Please set the JAVA_HOME variable in your environment to match the >&2
+echo location of your Java installation. >&2
+echo.
+goto error
+
+@REM ==== END VALIDATION ====
+
+:init
+
+@REM Find the project base dir, i.e. the directory that contains the folder ".mvn".
+@REM Fallback to current working directory if not found.
+
+set MAVEN_PROJECTBASEDIR=%MAVEN_BASEDIR%
+IF NOT "%MAVEN_PROJECTBASEDIR%"=="" goto endDetectBaseDir
+
+set EXEC_DIR=%CD%
+set WDIR=%EXEC_DIR%
+:findBaseDir
+IF EXIST "%WDIR%"\.mvn goto baseDirFound
+cd ..
+IF "%WDIR%"=="%CD%" goto baseDirNotFound
+set WDIR=%CD%
+goto findBaseDir
+
+:baseDirFound
+set MAVEN_PROJECTBASEDIR=%WDIR%
+cd "%EXEC_DIR%"
+goto endDetectBaseDir
+
+:baseDirNotFound
+set MAVEN_PROJECTBASEDIR=%EXEC_DIR%
+cd "%EXEC_DIR%"
+
+:endDetectBaseDir
+
+IF NOT EXIST "%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config" goto endReadAdditionalConfig
+
+@setlocal EnableExtensions EnableDelayedExpansion
+for /F "usebackq delims=" %%a in ("%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config") do set JVM_CONFIG_MAVEN_PROPS=!JVM_CONFIG_MAVEN_PROPS! %%a
+@endlocal & set JVM_CONFIG_MAVEN_PROPS=%JVM_CONFIG_MAVEN_PROPS%
+
+:endReadAdditionalConfig
+
+SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe"
+set WRAPPER_JAR="%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.jar"
+set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
+
+set DOWNLOAD_URL="https://repo.maven.apache.org/maven2/org/apache/maven/wrapper/maven-wrapper/3.1.0/maven-wrapper-3.1.0.jar"
+
+FOR /F "usebackq tokens=1,2 delims==" %%A IN ("%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.properties") DO (
+    IF "%%A"=="wrapperUrl" SET DOWNLOAD_URL=%%B
+)
+
+@REM Extension to allow automatically downloading the maven-wrapper.jar from Maven-central
+@REM This allows using the maven wrapper in projects that prohibit checking in binary data.
+if exist %WRAPPER_JAR% (
+    if "%MVNW_VERBOSE%" == "true" (
+        echo Found %WRAPPER_JAR%
+    )
+) else (
+    if not "%MVNW_REPOURL%" == "" (
+        SET DOWNLOAD_URL="%MVNW_REPOURL%/org/apache/maven/wrapper/maven-wrapper/3.1.0/maven-wrapper-3.1.0.jar"
+    )
+    if "%MVNW_VERBOSE%" == "true" (
+        echo Couldn't find %WRAPPER_JAR%, downloading it ...
+        echo Downloading from: %DOWNLOAD_URL%
+    )
+
+    powershell -Command "&{"^
+		"$webclient = new-object System.Net.WebClient;"^
+		"if (-not ([string]::IsNullOrEmpty('%MVNW_USERNAME%') -and [string]::IsNullOrEmpty('%MVNW_PASSWORD%'))) {"^
+		"$webclient.Credentials = new-object System.Net.NetworkCredential('%MVNW_USERNAME%', '%MVNW_PASSWORD%');"^
+		"}"^
+		"[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12; $webclient.DownloadFile('%DOWNLOAD_URL%', '%WRAPPER_JAR%')"^
+		"}"
+    if "%MVNW_VERBOSE%" == "true" (
+        echo Finished downloading %WRAPPER_JAR%
+    )
+)
+@REM End of extension
+
+@REM Provide a "standardized" way to retrieve the CLI args that will
+@REM work with both Windows and non-Windows executions.
+set MAVEN_CMD_LINE_ARGS=%*
+
+%MAVEN_JAVA_EXE% ^
+  %JVM_CONFIG_MAVEN_PROPS% ^
+  %MAVEN_OPTS% ^
+  %MAVEN_DEBUG_OPTS% ^
+  -classpath %WRAPPER_JAR% ^
+  "-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" ^
+  %WRAPPER_LAUNCHER% %MAVEN_CONFIG% %*
+if ERRORLEVEL 1 goto error
+goto end
+
+:error
+set ERROR_CODE=1
+
+:end
+@endlocal & set ERROR_CODE=%ERROR_CODE%
+
+if not "%MAVEN_SKIP_RC%"=="" goto skipRcPost
+@REM check for post script, once with legacy .bat ending and once with .cmd ending
+if exist "%USERPROFILE%\mavenrc_post.bat" call "%USERPROFILE%\mavenrc_post.bat"
+if exist "%USERPROFILE%\mavenrc_post.cmd" call "%USERPROFILE%\mavenrc_post.cmd"
+:skipRcPost
+
+@REM pause the script if MAVEN_BATCH_PAUSE is set to 'on'
+if "%MAVEN_BATCH_PAUSE%"=="on" pause
+
+if "%MAVEN_TERMINATE_CMD%"=="on" exit %ERROR_CODE%
+
+cmd /C exit /B %ERROR_CODE%
--- a/samples/sample-05/pom.xml
+++ b/samples/sample-05/pom.xml
@@ -0,0 +1,127 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-starter-parent</artifactId>
+		<version>3.0.0-SNAPSHOT</version>
+		<relativePath/> <!-- lookup parent from repository -->
+	</parent>
+
+	<groupId>com.example</groupId>
+	<artifactId>kafka-sample-05</artifactId>
+	<version>3.0.0-SNAPSHOT</version>
+
+	<name>kafka-sample-05</name>
+	<description>Kafka Sample 5</description>
+
+
+	<properties>
+		<java.version>17</java.version>
+		<spring-kafka.version>3.0.0-SNAPSHOT</spring-kafka.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka-clients</artifactId>
+			<classifier>test</classifier>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka_2.13</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka_2.13</artifactId>
+			<classifier>test</classifier>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.awaitility</groupId>
+			<artifactId>awaitility</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+			</plugin>
+			<plugin>
+				<artifactId>maven-surefire-plugin</artifactId>
+				<version>3.0.0-M7</version>
+				<configuration>
+					<properties>
+						<configurationParameters>
+							spring.kafka.global.embedded.enabled = true
+							spring.kafka.embedded.broker.properties.location = kafka-broker.properties
+						</configurationParameters>
+					</properties>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+
+	<repositories>
+		<repository>
+			<id>spring-milestones</id>
+			<name>Spring Milestones</name>
+			<url>https://repo.spring.io/milestone</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</repository>
+		<repository>
+			<id>spring-snapshots</id>
+			<name>Spring Snapshots</name>
+			<url>https://repo.spring.io/snapshot</url>
+			<releases>
+				<enabled>false</enabled>
+			</releases>
+		</repository>
+	</repositories>
+	<pluginRepositories>
+		<pluginRepository>
+			<id>spring-milestones</id>
+			<name>Spring Milestones</name>
+			<url>https://repo.spring.io/milestone</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</pluginRepository>
+		<pluginRepository>
+			<id>spring-snapshots</id>
+			<name>Spring Snapshots</name>
+			<url>https://repo.spring.io/snapshot</url>
+			<releases>
+				<enabled>false</enabled>
+			</releases>
+		</pluginRepository>
+	</pluginRepositories>
+
+</project>
--- a/samples/sample-05/src/main/java/com/example/Sample05Application.java
+++ b/samples/sample-05/src/main/java/com/example/Sample05Application.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.example;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.kafka.annotation.KafkaListener;
+
+/**
+ * Plain Kafka listener sample which purpose is only to be used in the tests assertions.
+ * See unit tests for this project for more information.
+ *
+ * @author Artem Bilan
+ *
+ * @since 3.0
+ *
+ */
+@SpringBootApplication
+public class Sample05Application {
+
+	public static void main(String[] args) {
+		SpringApplication.run(Sample05Application.class, args);
+	}
+
+	@KafkaListener(id ="sampleListener", topics = "topic1")
+	void listenForTopic(String payload) {
+		System.out.println("Received: " + payload);
+	}
+
+}
--- a/samples/sample-05/src/main/resources/application.properties
+++ b/samples/sample-05/src/main/resources/application.properties
@@ -0,0 +1,4 @@
+spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}
+spring.kafka.producer.properties[max.block.ms]=1000
+spring.kafka.consumer.auto-offset-reset=earliest
+logging.level.root=error
--- a/samples/sample-05/src/test/java/com/example/Sample05Application1Tests.java
+++ b/samples/sample-05/src/test/java/com/example/Sample05Application1Tests.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.example;
+
+import static org.awaitility.Awaitility.await;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.test.annotation.DirtiesContext;
+
+/**
+ * This test is going to fail from IDE since there is no exposed {@code spring.embedded.kafka.brokers} system property.
+ * Use Maven to run tests which enables global embedded Kafka broker via properties provided to Surefire plugin.
+ */
+@ExtendWith(OutputCaptureExtension.class)
+@SpringBootTest
+@DirtiesContext
+class Sample05Application1Tests {
+
+	@Autowired
+	KafkaTemplate<String, String> kafkaTemplate;
+
+	@Test
+	void testKafkaListener(CapturedOutput output) {
+		this.kafkaTemplate.send("topic1", "testData");
+
+		await().until(() -> output.getOut().contains("Received: testData"));
+	}
+
+}
--- a/samples/sample-05/src/test/java/com/example/Sample05Application2Tests.java
+++ b/samples/sample-05/src/test/java/com/example/Sample05Application2Tests.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.example;
+
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.kafka.common.errors.TimeoutException;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.KafkaException;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.test.annotation.DirtiesContext;
+
+/**
+ * This test is going to fail from IDE since there is no exposed {@code spring.embedded.kafka.brokers} system property.
+ * This test demonstrates that global embedded Kafka broker config for {@code auto.create.topics.enable=false}
+ * is in an effect - the topic {@code nonExistingTopic} does not exist on the broker.
+ * See {@code /resources/kafka-broker.properties} and Maven Surefire plugin configuration.
+ */
+@SpringBootTest
+@DirtiesContext
+class Sample05Application2Tests {
+
+	@Autowired
+	KafkaTemplate<String, String> kafkaTemplate;
+
+	@Test
+	void testKafkaTemplateSend() {
+		assertThatExceptionOfType(KafkaException.class)
+				.isThrownBy(() ->
+						this.kafkaTemplate.send("nonExistingTopic", "fake data").get(10, TimeUnit.SECONDS))
+				.withRootCauseExactlyInstanceOf(TimeoutException.class)
+				.withMessageContaining("Send failed")
+				.withStackTraceContaining("Topic nonExistingTopic not present in metadata");
+	}
+
+}
--- a/samples/sample-05/src/test/resources/junit-platform.properties
+++ b/samples/sample-05/src/test/resources/junit-platform.properties
@@ -0,0 +1 @@
+spring.kafka.embedded.topics=topic1,topic2
--- a/samples/sample-05/src/test/resources/kafka-broker.properties
+++ b/samples/sample-05/src/test/resources/kafka-broker.properties
@@ -0,0 +1 @@
+auto.create.topics.enable=false
--- a/samples/sample-05/src/test/resources/logback-test.xml
+++ b/samples/sample-05/src/test/resources/logback-test.xml
@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+
+	<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+		<!-- encoders are assigned the type
+			 ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+		<encoder>
+			<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+		</encoder>
+	</appender>
+
+	<logger name="org.springframework.kafka.test.junit" level="debug"/>
+	<logger name="org.springframework.kafka.test" level="warn"/>
+	<logger name="org.apache.kafka.clients" level="warn"/>
+	<logger name="org.apache.kafka.clients.NetworkClient" level="error"/>
+	<logger name="org.apache.kafka.common.network.Selector" level="error"/>
+	<logger name="kafka.server.ReplicaFetcherThread" level="error"/>
+	<logger name="kafka.server.LogDirFailureChannel" level="error"/>
+	<logger name="kafka.server.BrokerMetadataCheckpoint" level="error"/>
+	<logger name="kafka.utils.CoreUtils$" level="error"/>
+	<logger name="org.apache.zookeeper.server.ZooKeeperServer" level="error"/>
+
+	<root level="error">
+		<appender-ref ref="STDOUT" />
+	</root>
+</configuration>
--- a/spring-kafka-docs/src/main/asciidoc/testing.adoc
+++ b/spring-kafka-docs/src/main/asciidoc/testing.adoc
@@ -44,7 +44,7 @@ This is because, in most cases, you want the consumer to consume any messages se
 The `ConsumerConfig` default is `latest` which means that messages already sent by a test, before the consumer starts, will not receive those records.
 To revert to the previous behavior, set the property to `latest` after calling the method.

-When using the embedded broker, it is generally best practice to use a different topic for each test, to prevent cross-talk.
+When using the embedded broker, it is generally best practice using a different topic for each test, to prevent cross-talk.
 If this is not possible for some reason, note that the `consumeFromEmbeddedTopics` method's default behavior is to seek the assigned partitions to the beginning after assignment.
 Since it does not have access to the consumer properties, you must use the overloaded method that takes a `seekToEnd` boolean parameter to seek to the end instead of the beginning.
 ====
@@ -129,6 +129,11 @@ ConsumerRecord<Integer, String> received = KafkaTestUtils.getSingleRecord(consum
 When the embedded Kafka and embedded Zookeeper server are started by the `EmbeddedKafkaBroker`, a system property named `spring.embedded.kafka.brokers` is set to the address of the Kafka brokers and a system property named `spring.embedded.zookeeper.connect` is set to the address of Zookeeper.
 Convenient constants (`EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS` and `EmbeddedKafkaBroker.SPRING_EMBEDDED_ZOOKEEPER_CONNECT`) are provided for this property.

+Instead of default `spring.embedded.kafka.brokers` system property, the address of the Kafka brokers can be exposed to any arbitrary and convenient property.
+For this purpose a `spring.embedded.kafka.brokers.property` (`EmbeddedKafkaBroker.BROKER_LIST_PROPERTY`) system property can be set before starting an embedded Kafka.
+For example, with Spring Boot a `spring.kafka.bootstrap-servers` configuration property is expected to be set for auto-configuring Kafka client, respectively.
+So, before running tests with an embedded Kafka on random ports, we can set `spring.embedded.kafka.brokers.property=spring.kafka.bootstrap-servers` as a system property - and the `EmbeddedKafkaBroker`  will use it to expose its broker addresses.
+
 With the `EmbeddedKafkaBroker.brokerProperties(Map<String, String>)`, you can provide additional properties for the Kafka servers.
 See https://kafka.apache.org/documentation/#brokerconfigs[Kafka Config] for more information about possible broker properties.

@@ -158,9 +163,10 @@ public class MyTests {
 By default, `addTopics` will throw an exception when problems arise (such as adding a topic that already exists).
 Version 2.6 added a new version of that method that returns a `Map<String, Exception>`; the key is the topic name and the value is `null` for success, or an `Exception` for a failure.

+[[same-broker-multiple-tests]]
 ==== Using the Same Broker(s) for Multiple Test Classes

-There is no built-in support for doing so, but you can use the same broker for multiple test classes with something similar to the following:
+You can use the same broker for multiple test classes with something similar to the following:

 ====
 [source, java]
@@ -214,6 +220,25 @@ IMPORTANT: The preceding example provides no mechanism for shutting down the bro
 This could be a problem if, say, you run your tests in a Gradle daemon.
 You should not use this technique in such a situation, or you should use something to call `destroy()` on the `EmbeddedKafkaBroker` when your tests are complete.

+Starting with version 3.0, the framework exposes a `GlobalEmbeddedKafkaTestExecutionListener` for the JUnit Platform; it is disabled by default.
+The purpose of this listener is to start one global `EmbeddedKafkaBroker` for the whole test plan and stop it at the end of the plan.
+To enable this listener, and therefore have a single global embedded Kafka cluster for all the tests in the project, the `spring.kafka.global.embedded.enabled` property must be set to `true` via system properties or JUnit Platform configuration.
+In addition, these properties can be provided:
+
+- `spring.kafka.embedded.count` - the number of Kafka brokers to manage;
+- `spring.kafka.embedded.ports` - ports (comma-separated value) for every Kafka broker to start, `0` if random port is a preferred; the number of values must be equal to the `count` mentioned above;
+- `spring.kafka.embedded.topics` - topics (comma-separated value) to create in the started Kafka cluster;
+- `spring.kafka.embedded.partitions` - number of partitions to provision for the created topics;
+- `spring.kafka.embedded.broker.properties.location` - the location of the file for additional Kafka broker configuration properties; the value of this property must follow the Spring resource abstraction pattern.
+
+Essentially these properties mimic some of the `@EmbeddedKafka` attributes.
+
+See more information about configuration properties and how to provide them in the https://junit.org/junit5/docs/current/user-guide/#running-tests-config-params[JUnit 5 User Guide].
+For example, a `spring.embedded.kafka.brokers.property=spring.kafka.bootstrap-servers` entry (for testing in Spring Boot application) can be added into a `junit-platform.properties` file in the testing classpath.
+
+NOTE: It is recommended to not combine a global embedded Kafka and per-test class in a single test suite.
+Both of them share the same system properties, so it is very likely going to lead to unexpected behavior.
+
 [[embedded-kafka-annotation]]
 ==== @EmbeddedKafka Annotation
 We generally recommend that you use the rule as a `@ClassRule` to avoid starting and stopping the broker between tests (and use a different topic for each test).
--- a/spring-kafka-docs/src/main/asciidoc/whats-new.adoc
+++ b/spring-kafka-docs/src/main/asciidoc/whats-new.adoc
@@ -6,7 +6,7 @@ For changes in earlier version, see <<history>>.
 [[x30-kafka-client]]
 ==== Kafka Client Version

-This version requires the 3.1.0 `kafka-clients`
+This version requires the 3.2.0 `kafka-clients`.

 [[x30-eos]]
 ==== Exactly Once Semantics
@@ -16,3 +16,9 @@ This version requires the 3.1.0 `kafka-clients`
 IMPORTANT: When using transactions, the minimum broker version is 2.5.

 See <<exactly-once>> and https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+Producer+scalability+for+exactly+once+semantics[KIP-447] for more information.
+
+[[global-embedded-kafka]]
+==== Global Single Embedded Kafka
+
+The embedded Kafka (`EmbeddedKafkaBroker`) can now be start as a single global instance for the whole test plan.
+See <<same-broker-multiple-tests>> for more information.
--- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/junit/GlobalEmbeddedKafkaTestExecutionListener.java
+++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/junit/GlobalEmbeddedKafkaTestExecutionListener.java
@@ -0,0 +1,145 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.test.junit;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.platform.engine.ConfigurationParameters;
+import org.junit.platform.launcher.TestExecutionListener;
+import org.junit.platform.launcher.TestPlan;
+
+import org.springframework.core.io.DefaultResourceLoader;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.support.PropertiesLoaderUtils;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.util.StringUtils;
+
+/**
+ * The {@link TestExecutionListener} to start an {@link EmbeddedKafkaBroker}
+ * in the beginning of the test plan and stop in the end.
+ * This approach ensures one global Kafka cluster for all the unit tests to execute.
+ * <p>
+ * The {@link GlobalEmbeddedKafkaTestExecutionListener} is disabled by default.
+ * Set {@link GlobalEmbeddedKafkaTestExecutionListener#LISTENER_ENABLED_PROPERTY_NAME}
+ * system property (or respective {@link ConfigurationParameters#CONFIG_FILE_NAME} entry)
+ * to enable it.
+ *
+ * @author Artem Bilan
+ *
+ * @since 3.0
+ */
+public class GlobalEmbeddedKafkaTestExecutionListener implements TestExecutionListener {
+
+	private static final Log LOGGER = LogFactory.getLog(GlobalEmbeddedKafkaTestExecutionListener.class);
+
+	/**
+	 * Property name used to enable the {@code GlobalEmbeddedKafkaTestExecutionListener}.
+	 * The {@code GlobalEmbeddedKafkaTestExecutionListener} is registered automatically via
+	 * Java's {@link java.util.ServiceLoader} mechanism but disabled by default.
+	 * Set the value of this property to {@code true} to enable this listener.
+	 */
+	public static final String LISTENER_ENABLED_PROPERTY_NAME = "spring.kafka.global.embedded.enabled";
+
+	/**
+	 * The number of brokers for {@link EmbeddedKafkaBroker}.
+	 */
+	public static final String COUNT_PROPERTY_NAME = "spring.kafka.embedded.count";
+
+	/**
+	 * The port(s) to expose embedded broker(s).
+	 */
+	public static final String PORTS_PROPERTY_NAME = "spring.kafka.embedded.ports";
+
+	/**
+	 * The topics to create on the embedded broker(s).
+	 */
+	public static final String TOPICS_PROPERTY_NAME = "spring.kafka.embedded.topics";
+
+	/**
+	 * The number of partitions on topics to create on the embedded broker(s).
+	 */
+	public static final String PARTITIONS_PROPERTY_NAME = "spring.kafka.embedded.partitions";
+
+	/**
+	 * The location for a properties file with Kafka broker configuration.
+	 */
+	public static final String BROKER_PROPERTIES_LOCATION_PROPERTY_NAME =
+			"spring.kafka.embedded.broker.properties.location";
+
+	private EmbeddedKafkaBroker embeddedKafkaBroker;
+
+	@Override
+	public void testPlanExecutionStarted(TestPlan testPlan) {
+		ConfigurationParameters configurationParameters = testPlan.getConfigurationParameters();
+		boolean enabled = configurationParameters.getBoolean(LISTENER_ENABLED_PROPERTY_NAME).orElse(false);
+		if (enabled) {
+			Integer count = configurationParameters.get(COUNT_PROPERTY_NAME, Integer::parseInt).orElse(1);
+			String[] topics =
+					configurationParameters.get(TOPICS_PROPERTY_NAME, StringUtils::commaDelimitedListToStringArray)
+							.orElse(null);
+			Integer partitions = configurationParameters.get(PARTITIONS_PROPERTY_NAME, Integer::parseInt).orElse(2);
+			Map<String, String> brokerProperties =
+					configurationParameters.get(BROKER_PROPERTIES_LOCATION_PROPERTY_NAME, this::brokerProperties)
+							.orElse(null);
+			String brokerListProperty = configurationParameters.get(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY)
+					.orElse(null);
+			int[] ports =
+					configurationParameters.get(PORTS_PROPERTY_NAME, this::ports)
+							.orElse(new int[count]);
+
+			this.embeddedKafkaBroker =
+					new EmbeddedKafkaBroker(count, false, partitions, topics)
+							.brokerProperties(brokerProperties)
+							.brokerListProperty(brokerListProperty)
+							.kafkaPorts(ports);
+			this.embeddedKafkaBroker.afterPropertiesSet();
+
+			LOGGER.info("Started global Embedded Kafka on: " + this.embeddedKafkaBroker.getBrokersAsString());
+		}
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	private Map<String, String> brokerProperties(String propertiesLocation) {
+		Resource propertiesResource = new DefaultResourceLoader().getResource(propertiesLocation);
+		try {
+			return (Map) PropertiesLoaderUtils.loadProperties(propertiesResource);
+		}
+		catch (IOException ex) {
+			throw new UncheckedIOException(ex);
+		}
+	}
+
+	private int[] ports(String ports) {
+		return StringUtils.commaDelimitedListToSet(ports)
+				.stream()
+				.mapToInt(Integer::parseInt)
+				.toArray();
+	}
+
+	@Override
+	public void testPlanExecutionFinished(TestPlan testPlan) {
+		if (this.embeddedKafkaBroker != null) {
+			this.embeddedKafkaBroker.destroy();
+			LOGGER.info("Stopped global Embedded Kafka.");
+		}
+	}
+
+}
--- a/spring-kafka-test/src/main/resources/META-INF/services/org.junit.platform.launcher.TestExecutionListener
+++ b/spring-kafka-test/src/main/resources/META-INF/services/org.junit.platform.launcher.TestExecutionListener
@@ -0,0 +1 @@
+org.springframework.kafka.test.junit.GlobalEmbeddedKafkaTestExecutionListener
--- a/spring-kafka-test/src/test/java/org/springframework/kafka/test/condition/WithSpringTestContextTests.java
+++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/condition/WithSpringTestContextTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
@@ -24,6 +24,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;

 /**
@@ -33,6 +34,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 */
@SpringJUnitConfig
@EmbeddedKafka
+@DirtiesContext
 public class WithSpringTestContextTests {

 	@Test
--- a/spring-kafka-test/src/test/java/org/springframework/kafka/test/junit/GlobalEmbeddedKafkaTestExecutionListenerTests.java
+++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/junit/GlobalEmbeddedKafkaTestExecutionListenerTests.java
@@ -0,0 +1,151 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.test.junit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.platform.engine.discovery.DiscoverySelectors;
+import org.junit.platform.launcher.core.LauncherDiscoveryRequestBuilder;
+import org.junit.platform.launcher.core.LauncherFactory;
+import org.junit.platform.launcher.listeners.SummaryGeneratingListener;
+
+import org.springframework.util.DefaultPropertiesPersister;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 3.0
+ */
+public class GlobalEmbeddedKafkaTestExecutionListenerTests {
+
+	@BeforeAll
+	static void setup() {
+		System.setProperty(GlobalEmbeddedKafkaTestExecutionListener.LISTENER_ENABLED_PROPERTY_NAME, "true");
+	}
+
+	@AfterAll
+	static void tearDown() {
+		System.clearProperty(GlobalEmbeddedKafkaTestExecutionListener.LISTENER_ENABLED_PROPERTY_NAME);
+		System.clearProperty(GlobalEmbeddedKafkaTestExecutionListener.BROKER_PROPERTIES_LOCATION_PROPERTY_NAME);
+	}
+
+	@Test
+	void testGlobalEmbeddedKafkaTestExecutionListener() throws IOException {
+		var brokerProperties = new Properties();
+		brokerProperties.setProperty("auto.create.topics.enable", "false");
+
+		var propertiesFile = File.createTempFile("kafka-broker", ".properties");
+
+		try (var outputStream = new BufferedOutputStream(new FileOutputStream(propertiesFile))) {
+			new DefaultPropertiesPersister().store(brokerProperties, outputStream, "Last entry");
+		}
+
+		System.setProperty(GlobalEmbeddedKafkaTestExecutionListener.BROKER_PROPERTIES_LOCATION_PROPERTY_NAME,
+				"file:" + propertiesFile.getAbsolutePath());
+
+		var discoveryRequest =
+				LauncherDiscoveryRequestBuilder.request()
+						.selectors(DiscoverySelectors.selectClass(TestClass1.class),
+								DiscoverySelectors.selectClass(TestClass2.class))
+						.build();
+
+		var summaryGeneratingListener = new SummaryGeneratingListener();
+		LauncherFactory.create().execute(discoveryRequest, summaryGeneratingListener);
+
+		var summary = summaryGeneratingListener.getSummary();
+
+		try {
+			assertThat(summary.getTestsStartedCount()).isEqualTo(2);
+			assertThat(summary.getTestsSucceededCount()).isEqualTo(1);
+			assertThat(summary.getTestsFailedCount()).isEqualTo(1);
+		}
+		catch (Exception ex) {
+			summary.printFailuresTo(new PrintWriter(System.out));
+			throw ex;
+		}
+	}
+
+	@EnabledIfSystemProperty(named = GlobalEmbeddedKafkaTestExecutionListener.LISTENER_ENABLED_PROPERTY_NAME,
+			matches = "true")
+	static class TestClass1 {
+
+		@Test
+		void testDescribeTopic() throws ExecutionException, InterruptedException, TimeoutException {
+			Map<String, Object> adminConfigs =
+					Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
+							System.getProperty("spring.kafka.bootstrap-servers"));
+			try (var admin = AdminClient.create(adminConfigs)) {
+				var topicsMap =
+						admin.describeTopics(Set.of("topic1", "topic2"))
+								.allTopicNames()
+								.get(10, TimeUnit.SECONDS);
+
+				assertThat(topicsMap).containsOnlyKeys("topic1", "topic2");
+			}
+		}
+
+	}
+
+	@EnabledIfSystemProperty(named = GlobalEmbeddedKafkaTestExecutionListener.LISTENER_ENABLED_PROPERTY_NAME,
+			matches = "true")
+	static class TestClass2 {
+
+		@Test
+		void testCannotAutoCreateTopic() throws ExecutionException, InterruptedException, TimeoutException {
+			Map<String, Object> producerConfigs = new HashMap<>();
+			producerConfigs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
+					System.getProperty("spring.kafka.bootstrap-servers"));
+			producerConfigs.put(ProducerConfig.RETRIES_CONFIG, 1);
+			producerConfigs.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 1);
+			producerConfigs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10);
+
+			StringSerializer serializer = new StringSerializer();
+			try (var kafkaProducer = new KafkaProducer<>(producerConfigs, serializer, serializer)) {
+				var recordMetadata =
+						kafkaProducer.send(new ProducerRecord<>("nonExistingTopic", "testValue"))
+								.get(10, TimeUnit.SECONDS);
+
+				assertThat(recordMetadata).isNotNull();
+			}
+		}
+
+	}
+
+}
--- a/spring-kafka-test/src/test/java/org/springframework/kafka/test/rule/AddressableEmbeddedBrokerTests.java
+++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/rule/AddressableEmbeddedBrokerTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2018-2022 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
@@ -36,6 +36,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;

 /**
@@ -48,6 +49,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 *
 */
@SpringJUnitConfig
+@DirtiesContext
 public class AddressableEmbeddedBrokerTests {

 	private static final String TEST_EMBEDDED = "testEmbedded";
--- a/spring-kafka-test/src/test/resources/junit-platform.properties
+++ b/spring-kafka-test/src/test/resources/junit-platform.properties
@@ -0,0 +1,3 @@
+spring.kafka.embedded.count=2
+spring.embedded.kafka.brokers.property=spring.kafka.bootstrap-servers
+spring.kafka.embedded.topics=topic1,topic2
				`@@ -0,0 +1 @@`
				`org.springframework.kafka.test.junit.GlobalEmbeddedKafkaTestExecutionListener`