Log TX Exceptions

- Log exceptions on commit/abort transaction
- Don't attempt to abort if commit fails, otherwise we see

`Abort failed`
`org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state`

**cherry-pick to 2.1.x**

* Rename variable to detect whether the commit failed

* Polishing; use try/catch around commit

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

Author: garyrussell

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

@@ -417,6 +417,7 @@ public void commitTransaction() throws ProducerFencedException {
 				this.delegate.commitTransaction();
 			}
 			catch (RuntimeException e) {
+				logger.error("Commit failed", e);
 				this.txFailed = true;
 				throw e;
 			}
@@ -428,6 +429,7 @@ public void abortTransaction() throws ProducerFencedException {
 				this.delegate.abortTransaction();
 			}
 			catch (RuntimeException e) {
+				logger.error("Abort failed", e);
 				this.txFailed = true;
 				throw e;
 			}

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

@@ -271,7 +271,17 @@ public <T> T executeInTransaction(OperationsCallback<K, V, T> callback) {
 		this.producers.set(producer);
 		T result = null;
 		try {
-			result = callback.doInOperations(this);
+			T result = callback.doInOperations(this);
+			try {
+				producer.commitTransaction();
+			}
+			catch (Exception e) {
+				throw new SkipAbortException(e);
+			}
+			return result;
+		}
+		catch (SkipAbortException e) {
+			throw ((RuntimeException) e.getCause());
 		}
 		catch (Exception e) {
 			try {
@@ -415,4 +425,13 @@ private Producer<K, V> getTheProducer() {
 		}
 	}
 
+	@SuppressWarnings("serial")
+	private static final class SkipAbortException extends RuntimeException {
+
+		SkipAbortException(Throwable cause) {
+			super(cause);
+		}
+
+	}
+
 }