Polishing JavaDocs and possible NPEs

https://build.spring.io/browse/INTEXT-KAFKA-310/
spring-attic · Aug 29, 2019 · a39c2b6 · a39c2b6
1 parent 90108c1
commit a39c2b6
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 22 deletions.
diff --git a/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java b/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java
@@ -101,7 +101,7 @@ public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
 	 * @return the spec.
 	 * @since 3.0.1
 	 * @deprecated in favor of
-	 * {@link #inboundChannelAdapter(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory)}
+	 * {@code #inboundChannelAdapter(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory)}
 	 */
 	@Deprecated
 	public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
@@ -123,7 +123,7 @@ public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
 	 * @return the spec.
 	 * @since 3.0.1
 	 * @deprecated in favor of
-	 * {@link #inboundChannelAdapter(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean)}
+	 * {@code #inboundChannelAdapter(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean)}
 	 */
 	@Deprecated
 	public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
@@ -132,7 +132,8 @@ public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
 			boolean allowMultiFetch,
 			String... topics) {
 
-		return new KafkaInboundChannelAdapterSpec<>(consumerFactory, new ConsumerProperties(topics), ackCallbackFactory, allowMultiFetch);
+		return new KafkaInboundChannelAdapterSpec<>(consumerFactory, new ConsumerProperties(topics),
+				ackCallbackFactory, allowMultiFetch);
 	}
 
 	/**

diff --git a/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
@@ -49,7 +49,6 @@
 import org.springframework.integration.acks.AcknowledgmentCallbackFactory;
 import org.springframework.integration.core.Pausable;
 import org.springframework.integration.endpoint.AbstractMessageSource;
-import org.springframework.integration.kafka.inbound.KafkaMessageSource.KafkaAckInfo;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
@@ -251,8 +250,8 @@ public KafkaMessageSource(ConsumerFactory<K, V> consumerFactory,
 		Assert.notNull(ackCallbackFactory, "'ackCallbackFactory' must not be null");
 		Assert.isTrue(
 				!ObjectUtils.isEmpty(consumerProperties.getTopics())
-				|| !ObjectUtils.isEmpty(consumerProperties.getTopicPartitionsToAssign())
-				|| consumerProperties.getTopicPattern() != null,
+						|| !ObjectUtils.isEmpty(consumerProperties.getTopicPartitionsToAssign())
+						|| consumerProperties.getTopicPattern() != null,
 				"topics, topicPattern, or topicPartitions must be provided"
 		);
 		this.consumerProperties = consumerProperties;
@@ -623,7 +622,8 @@ else if (partition.isRelativeToCurrent()) {
 							}
 							catch (Exception e) {
 								this.logger.error("Failed to set initial offset for " + topicPartition
-										+ " at " + newOffset + ". Position is " + this.consumer.position(topicPartition), e);
+										+ " at " + newOffset + ". Position is " + this.consumer
+										.position(topicPartition), e);
 							}
 						}
 					}
@@ -678,7 +678,7 @@ public static class KafkaAckCallbackFactory<K, V> implements AcknowledgmentCallb
 		/**
 		 * Deprecated constructor.
 		 * @deprecated in favor of
-		 * {@link #KafkaMessageSource$KafkaAckCallbackFactory(ConsumerProperties)}.
+		 * {@code #KafkaAckCallbackFactory(ConsumerProperties)}.
 		 */
 		@Deprecated
 		public KafkaAckCallbackFactory() {
@@ -691,9 +691,8 @@ public KafkaAckCallbackFactory(ConsumerProperties consumerProperties) {
 
 		/**
 		 * Deprecated setter.
-		 * @deprecated in favor of
-		 * {@link #KafkaMessageSource$KafkaAckCallbackFactory(ConsumerProperties)}.
 		 * @param commitTimeout the commit timeout.
+		 * @deprecated in favor of {@code #KafkaAckCallbackFactory(ConsumerProperties)}.
 		 */
 		@Deprecated
 		public void setCommitTimeout(Duration commitTimeout) {
@@ -734,8 +733,7 @@ public static class KafkaAckCallback<K, V> implements AcknowledgmentCallback, Ac
 		/**
 		 * Deprecated constructor.
 		 * @param ackInfo the ack info.
-		 * @deprecated in favor of
-		 * {@link #KafkaMessageSource$KafkaAckCallback(KafkaAckInfo, ConsumerProperties)}
+		 * @deprecated in favor of {@code #KafkaAckCallback(KafkaAckInfo, ConsumerProperties)}
 		 */
 		@Deprecated
 		public KafkaAckCallback(KafkaAckInfo<K, V> ackInfo) {
@@ -746,8 +744,7 @@ public KafkaAckCallback(KafkaAckInfo<K, V> ackInfo) {
 		 * Deprecated constructor.
 		 * @param ackInfo the ack info.
 		 * @param commitTimeout the commit timeout.
-		 * @deprecated in favor of
-		 * {@link #KafkaMessageSource4KafkaAckCallback(KafkaAckInfo, ConsumerProperties)}
+		 * @deprecated in favor of {@code #KafkaAckCallback(KafkaAckInfo, ConsumerProperties)}
 		 */
 		@Deprecated
 		public KafkaAckCallback(KafkaAckInfo<K, V> ackInfo, @Nullable Duration commitTimeout) {
@@ -761,16 +758,18 @@ public KafkaAckCallback(KafkaAckInfo<K, V> ackInfo, @Nullable Duration commitTim
 		 * properties are used.
 		 */
 		public KafkaAckCallback(KafkaAckInfo<K, V> ackInfo, @Nullable ConsumerProperties consumerProperties) {
-
-				Assert.notNull(ackInfo, "'ackInfo' cannot be null");
+			Assert.notNull(ackInfo, "'ackInfo' cannot be null");
 			this.ackInfo = ackInfo;
-			this.commitTimeout = consumerProperties.getSyncCommitTimeout();
-			this.isSyncCommits = consumerProperties == null ? true : consumerProperties.isSyncCommits();
-			this.commitCallback = consumerProperties != null && consumerProperties.getCommitCallback() != null
-					? consumerProperties.getCommitCallback()
-					: new LoggingCommitCallback();
+			this.commitTimeout = consumerProperties != null ? consumerProperties.getSyncCommitTimeout() : null;
+			this.isSyncCommits = consumerProperties == null || consumerProperties.isSyncCommits();
+			this.commitCallback =
+					consumerProperties != null && consumerProperties.getCommitCallback() != null
+							? consumerProperties.getCommitCallback()
+							: new LoggingCommitCallback();
 			this.commitLogger = new LogIfLevelEnabled(this.logger,
-					consumerProperties.getCommitLogLevel());
+					consumerProperties != null
+							? consumerProperties.getCommitLogLevel()
+							: LogIfLevelEnabled.Level.DEBUG);
 		}
 
 		@Override