Fix the previous commit

https://build.spring.io/browse/SK-SK21X-59

Author: artembilan

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -255,6 +255,7 @@ public <T> T execute(ProducerCallback<K, V, T> callback) {
 
 	@Override
 	public <T> T executeInTransaction(OperationsCallback<K, V, T> callback) {
+		Assert.notNull(callback, "'callback' cannot be null");
 		Assert.state(this.transactional, "Producer factory does not support transactions");
 		Producer<K, V> producer = this.producers.get();
 		Assert.state(producer == null, "Nested calls to 'executeInTransaction' are not allowed");
@@ -269,7 +270,6 @@ public <T> T executeInTransaction(OperationsCallback<K, V, T> callback) {
 		}
 
 		this.producers.set(producer);
-		T result = null;
 		try {
 			T result = callback.doInOperations(this);
 			try {
@@ -284,23 +284,13 @@ public <T> T executeInTransaction(OperationsCallback<K, V, T> callback) {
 			throw ((RuntimeException) e.getCause());
 		}
 		catch (Exception e) {
-			try {
-				producer.abortTransaction();
-			}
-			finally {
-				this.producers.remove();
-				closeProducer(producer, false);
-			}
+			producer.abortTransaction();
 			throw e;
 		}
-		try {
-			producer.commitTransaction();
-		}
 		finally {
 			this.producers.remove();
 			closeProducer(producer, false);
 		}
-		return result;
 	}
 
 	/**