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

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

@@ -466,6 +466,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 				this.delegate.commitTransaction();
 			}
 			catch (RuntimeException e) {
+				logger.error("Commit failed", e);
 				this.txFailed = true;
 				throw e;
 			}
@@ -477,6 +478,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 				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

@@ -276,9 +276,17 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 		this.producers.set(producer);
 		try {
 			T result = callback.doInOperations(this);
-			producer.commitTransaction();
+			try {
+				producer.commitTransaction();
+			}
+			catch (Exception e) {
+				throw new SkipAbortException(e);
+			}
 			return result;
 		}
+		catch (SkipAbortException e) {
+			throw ((RuntimeException) e.getCause());
+		}
 		catch (Exception e) {
 			producer.abortTransaction();
 			throw e;
@@ -411,4 +419,13 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 		}
 	}
 
+	@SuppressWarnings("serial")
+	private static final class SkipAbortException extends RuntimeException {
+
+		SkipAbortException(Throwable cause) {
+			super(cause);
+		}
+
+	}
+
 }