From f6d858c7aad76656c536245b268eb07ee68a92dd Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 29 Aug 2019 19:08:15 -0400 Subject: [PATCH] Polishing JavaDocs and possible NPEs https://build.spring.io/browse/INTEXT-KAFKA-310/ --- .../integration/kafka/dsl/Kafka.java | 7 ++-- .../kafka/inbound/KafkaMessageSource.java | 37 +++++++++---------- 2 files changed, 22 insertions(+), 22 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java index a8c5b806ae..114a9fd33a 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java @@ -101,7 +101,7 @@ public final class Kafka { * @return the spec. * @since 3.0.1 * @deprecated in favor of - * {@link #inboundChannelAdapter(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory)} + * {@code #inboundChannelAdapter(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory)} */ @Deprecated public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( @@ -123,7 +123,7 @@ public final class Kafka { * @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 KafkaInboundChannelAdapterSpec inboundChannelAdapter( @@ -132,7 +132,8 @@ public final class Kafka { 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/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 410d084196..6e5fdde330 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 @@ -49,7 +49,6 @@ import org.springframework.integration.acks.AcknowledgmentCallback; 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 class KafkaMessageSource extends AbstractMessageSource impl 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 @@ public class KafkaMessageSource extends AbstractMessageSource impl } 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 class KafkaMessageSource extends AbstractMessageSource impl /** * Deprecated constructor. * @deprecated in favor of - * {@link #KafkaMessageSource$KafkaAckCallbackFactory(ConsumerProperties)}. + * {@code #KafkaAckCallbackFactory(ConsumerProperties)}. */ @Deprecated public KafkaAckCallbackFactory() { @@ -691,9 +691,8 @@ public class KafkaMessageSource extends AbstractMessageSource impl /** * 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 class KafkaMessageSource extends AbstractMessageSource impl /** * 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 ackInfo) { @@ -746,8 +744,7 @@ public class KafkaMessageSource extends AbstractMessageSource impl * 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 ackInfo, @Nullable Duration commitTimeout) { @@ -761,16 +758,18 @@ public class KafkaMessageSource extends AbstractMessageSource impl * properties are used. */ public KafkaAckCallback(KafkaAckInfo 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