GH-1672: Sonar Fix (Complexity)

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -1829,22 +1829,7 @@ private void invokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
 						TransactionSupport
 								.setTransactionIdSuffix(zombieFenceTxIdSuffix(record.topic(), record.partition()));
 					}
-					this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
-
-						@Override
-						public void doInTransactionWithoutResult(TransactionStatus s) {
-							if (ListenerConsumer.this.kafkaTxManager != null) {
-								ListenerConsumer.this.producer = ((KafkaResourceHolder) TransactionSynchronizationManager
-										.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
-												.getProducer(); // NOSONAR
-							}
-							RuntimeException aborted = doInvokeRecordListener(record, iterator);
-							if (aborted != null) {
-								throw aborted;
-							}
-						}
-
-					});
+					invokeInTransaction(iterator, record);
 				}
 				catch (ProducerFencedException | FencedInstanceIdException e) {
 					this.logger.error(e, "Producer or 'group.instance.id' fenced during transaction");
@@ -1870,6 +1855,25 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 			}
 		}
 
+		private void invokeInTransaction(Iterator<ConsumerRecord<K, V>> iterator, final ConsumerRecord<K, V> record) {
+			this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
+
+				@Override
+				public void doInTransactionWithoutResult(TransactionStatus s) {
+					if (ListenerConsumer.this.kafkaTxManager != null) {
+						ListenerConsumer.this.producer = ((KafkaResourceHolder) TransactionSynchronizationManager
+								.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
+										.getProducer(); // NOSONAR
+					}
+					RuntimeException aborted = doInvokeRecordListener(record, iterator);
+					if (aborted != null) {
+						throw aborted;
+					}
+				}
+
+			});
+		}
+
 		private void recordAfterRollback(Iterator<ConsumerRecord<K, V>> iterator, final ConsumerRecord<K, V> record,
 				RuntimeException e) {