Fix new Sonar smells

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();
+		}
+
+	}
+
 }

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.