From 094eb2e021f413d6072fb7ce79455fac9dd9751c Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@vmware.com>
Date: Thu, 6 Aug 2020 10:45:44 -0400
Subject: [PATCH] Fix new Sonar smells

---
 .../endpoint/ReactiveStreamsConsumer.java     | 56 +++++++++++--------
 .../kafka/inbound/KafkaMessageSource.java     |  2 +-
 2 files changed, 35 insertions(+), 23 deletions(-)

diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
index 470becba90..0b7901c25d 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
@@ -173,28 +173,7 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
 		else if (this.subscriber != null) {
 			this.subscription =
 					Flux.from(this.publisher)
-							.subscribeWith(new BaseSubscriber<Message<?>>() {
-
-								@Override
-								protected void hookOnSubscribe(Subscription subscription) {
-									ReactiveStreamsConsumer.this.subscriber.onSubscribe(subscription);
-								}
-
-								@Override
-								protected void hookOnNext(Message<?> value) {
-									try {
-										ReactiveStreamsConsumer.this.subscriber.onNext(value);
-									}
-									catch (Exception ex) {
-										ReactiveStreamsConsumer.this.errorHandler.handleError(ex);
-									}
-								}
-
-								@Override
-								protected void hookOnComplete() {
-									ReactiveStreamsConsumer.this.subscriber.onComplete();
-								}
-							});
+							.subscribeWith(new SubscriberDecorator(this.subscriber, this.errorHandler));
 		}
 	}
 
@@ -279,4 +258,37 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
 
 	}
 
+	private static final class SubscriberDecorator extends BaseSubscriber<Message<?>> {
+
+		private final Subscriber<Message<?>> delegate;
+
+		private final ErrorHandler errorHandler;
+
+		SubscriberDecorator(Subscriber<Message<?>> delegate, ErrorHandler errorHandler) {
+			this.delegate = delegate;
+			this.errorHandler = errorHandler;
+		}
+
+		@Override
+		protected void hookOnSubscribe(Subscription subscription) {
+			this.delegate.onSubscribe(subscription);
+		}
+
+		@Override
+		protected void hookOnNext(Message<?> value) {
+			try {
+				this.delegate.onNext(value);
+			}
+			catch (Exception ex) {
+				this.errorHandler.handleError(ex);
+			}
+		}
+
+		@Override
+		protected void hookOnComplete() {
+			this.delegate.onComplete();
+		}
+
+	}
+
 }
diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
index 87ce8801dd..4a13f09d25 100644
--- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
+++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
@@ -101,7 +101,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 
 	private static final long MIN_ASSIGN_TIMEOUT = 2000L;
 
-	private static int DEFAULT_CLOSE_TIMEOUT = 30;
+	private static final int DEFAULT_CLOSE_TIMEOUT = 30;
 
 	/**
 	 * The number of records remaining from the previous poll.