Manage log.dir in the EmbeddedKafkaBroker

Resolves spring-projects#194

Create the temporary directory in EKB instead of the broker to avoid
`NoSuchFileException`s during shutdown.

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

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

@@ -18,7 +18,10 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.net.InetSocketAddress;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -136,6 +139,8 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
 
 	private volatile ZooKeeperClient zooKeeperClient;
 
+	private String logDir;
+
 	public EmbeddedKafkaBroker(int count) {
 		this(count, false);
 	}
@@ -279,6 +284,7 @@ public synchronized EmbeddedKafkaBroker zkSessionTimeout(int zkSessionTimeout) {
 
 	@Override
 	public void afterPropertiesSet() {
+		logDir();
 		overrideExitMethods();
 		try {
 			this.zookeeper = new EmbeddedZookeeper(this.zkPort);
@@ -316,6 +322,19 @@ public void afterPropertiesSet() {
 		System.setProperty(SPRING_EMBEDDED_ZOOKEEPER_CONNECT, getZookeeperConnectionString());
 	}
 
+	private void logDir() {
+		if (this.brokerProperties.get(KafkaConfig.LogDirProp()) == null) {
+			try {
+				Path dir = Files.createTempDirectory("spring.kafka" + System.currentTimeMillis());
+				this.logDir = dir.toString();
+				this.brokerProperties.put(KafkaConfig.LogDirProp(), this.logDir);
+			}
+			catch (IOException e) {
+				throw new UncheckedIOException(e);
+			}
+		}
+	}
+
 	private void overrideExitMethods() {
 		String exitMsg = "Exit.%s(%d, %s) called";
 		Exit.setExitProcedure((statusCode, message) -> {