More test diagnostics

Author: garyrussell

spring-kafka/src/test/java/org/springframework/kafka/core/reactive/ReactiveKafkaProducerTemplateTransactionIntegrationTests.java

@@ -57,6 +57,7 @@
 import reactor.kafka.sender.SenderOptions;
 import reactor.kafka.sender.SenderRecord;
 import reactor.kafka.sender.SenderResult;
+import reactor.kafka.sender.internals.DefaultKafkaSender;
 import reactor.test.StepVerifier;
 
 /**
@@ -281,6 +282,7 @@ public void shouldSendOneRecordTransactionallyViaTemplateAsSenderRecordAndReceiv
 
 	@LogLevels(categories = "reactor.kafka.receiver.internals.ConsumerEventLoop", level = "TRACE",
 			classes = { JUnitUtils.class, LogLevelsCondition.class,
+					DefaultKafkaSender.class,
 					ReactiveKafkaProducerTemplateTransactionIntegrationTests.class})
 	@Test
 	public void shouldSendOneRecordTransactionallyViaTemplateAsSenderRecordAndReceiveItExactlyOnce() {
@@ -293,21 +295,20 @@ public void shouldSendOneRecordTransactionallyViaTemplateAsSenderRecordAndReceiv
 				.expectComplete()
 				.verify();
 
-		StepVerifier.create(reactiveKafkaConsumerTemplate
-				.receiveExactlyOnce(this.reactiveKafkaProducerTemplate.transactionManager())
-				.concatMap(consumerRecordFlux -> sendAndCommit(consumerRecordFlux, false))
-				.onErrorResume(error -> this.reactiveKafkaProducerTemplate.transactionManager()
-						.abort()
-						.then(Mono.error(error))))
+		StepVerifier.create(
+				reactiveKafkaConsumerTemplate
+						.receiveExactlyOnce(this.reactiveKafkaProducerTemplate.transactionManager())
+						.concatMap(consumerRecordFlux -> sendAndCommit(consumerRecordFlux, false)))
 				.assertNext(senderResult -> {
 					assertThat(senderResult.correlationMetadata().intValue()).isEqualTo(DEFAULT_KEY);
 					assertThat(senderResult.recordMetadata().offset()).isGreaterThan(0);
 				})
 				.thenCancel()
 				.verify(DEFAULT_VERIFY_TIMEOUT);
 
-		StepVerifier.create(reactiveKafkaConsumerTemplate
-				.receive().doOnNext(receiverRecord -> receiverRecord.receiverOffset().acknowledge()))
+		StepVerifier.create(
+				reactiveKafkaConsumerTemplate
+						.receive().doOnNext(receiverRecord -> receiverRecord.receiverOffset().acknowledge()))
 				.assertNext(receiverRecord -> {
 					logger.info(ListenerUtils.recordToString(receiverRecord, true));
 					assertThat(receiverRecord.value()).isEqualTo(DEFAULT_VALUE + "xyz");