Prevent EmbeddedKafkaBroker exiting the JVM

See #194
See #345
See gradle/gradle#11195

**cherry-pick to 2.4.x, 2.3.x, 2.2.x, 1.3.x**

# Conflicts:
#	spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java

Author: garyrussell

spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java

@@ -46,6 +46,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.utils.Exit;
 import org.apache.kafka.common.utils.Time;
 
 import org.springframework.beans.factory.DisposableBean;
@@ -202,6 +203,7 @@ public void setAdminTimeout(int adminTimeout) {
 
 	@Override
 	public void afterPropertiesSet() {
+		overrideExitMethods();
 		this.zookeeper = new EmbeddedZookeeper();
 		int zkConnectionTimeout = 6000; // NOSONAR magic #
 		int zkSessionTimeout = 6000; // NOSONAR magic #
@@ -235,6 +237,26 @@ public void afterPropertiesSet() {
 		System.setProperty(SPRING_EMBEDDED_ZOOKEEPER_CONNECT, getZookeeperConnectionString());
 	}
 
+	private void overrideExitMethods() {
+		String exitMsg = "Exit.%s(%d, %s) called";
+		Exit.setExitProcedure((statusCode, message) -> {
+			if (logger.isDebugEnabled()) {
+				logger.debug(new RuntimeException(), String.format(exitMsg, "exit", statusCode, message));
+			}
+			else {
+				logger.warn(String.format(exitMsg, "exit", statusCode, message));
+			}
+		});
+		Exit.setHaltProcedure((statusCode, message) -> {
+			if (logger.isDebugEnabled()) {
+				logger.debug(new RuntimeException(), String.format(exitMsg, "halt", statusCode, message));
+			}
+			else {
+				logger.warn(String.format(exitMsg, "halt", statusCode, message));
+			}
+		});
+	}
+
 	private Properties createBrokerProperties(int i) {
 		return TestUtils.createBrokerConfig(i, this.zkConnect, this.controlledShutdown,
 				true, this.kafkaPorts[i],