diff --git a/pom.xml b/pom.xml index ff62ab0be..800c469e3 100644 --- a/pom.xml +++ b/pom.xml @@ -108,27 +108,6 @@ - - org.apache.kafka - kafka_2.11 - test - test - ${kafka.version} - - - jline - jline - - - org.slf4j - slf4j-log4j12 - - - log4j - log4j - - - org.springframework.cloud spring-cloud-stream-schema diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/JaasLoginModuleConfiguration.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/JaasLoginModuleConfiguration.java index aa4e4755c..e249a1868 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/JaasLoginModuleConfiguration.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/JaasLoginModuleConfiguration.java @@ -37,10 +37,10 @@ public class JaasLoginModuleConfiguration { private KafkaJaasLoginModuleInitializer.ControlFlag controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag.REQUIRED; - private Map options = new HashMap<>(); + private Map options = new HashMap<>(); public String getLoginModule() { - return loginModule; + return this.loginModule; } public void setLoginModule(String loginModule) { @@ -49,7 +49,7 @@ public class JaasLoginModuleConfiguration { } public KafkaJaasLoginModuleInitializer.ControlFlag getControlFlag() { - return controlFlag; + return this.controlFlag; } public void setControlFlag(String controlFlag) { @@ -58,7 +58,7 @@ public class JaasLoginModuleConfiguration { } public Map getOptions() { - return options; + return this.options; } public void setOptions(Map options) { diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index decce7142..28cb3b079 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java @@ -40,6 +40,9 @@ import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; /** + * Configuration properties for the Kafka binder. + * The properties in this class are prefixed with spring.cloud.stream.kafka.binder. + * * @author David Turanski * @author Ilayaperumal Gopinathan * @author Marius Bogoevici @@ -134,7 +137,7 @@ public class KafkaBinderConfigurationProperties { } public KafkaProperties getKafkaProperties() { - return kafkaProperties; + return this.kafkaProperties; } public Transaction getTransaction() { @@ -167,7 +170,7 @@ public class KafkaBinderConfigurationProperties { /** * No longer used. * @return the window. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0") @@ -178,7 +181,7 @@ public class KafkaBinderConfigurationProperties { /** * No longer used. * @return the count. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0") @@ -189,7 +192,7 @@ public class KafkaBinderConfigurationProperties { /** * No longer used. * @return the timeout. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0") @@ -249,7 +252,7 @@ public class KafkaBinderConfigurationProperties { /** * No longer used. * @param offsetUpdateTimeWindow the window. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0") @@ -260,7 +263,7 @@ public class KafkaBinderConfigurationProperties { /** * No longer used. * @param offsetUpdateCount the count. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0") @@ -271,7 +274,7 @@ public class KafkaBinderConfigurationProperties { /** * No longer used. * @param offsetUpdateShutdownTimeout the timeout. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0") @@ -325,8 +328,11 @@ public class KafkaBinderConfigurationProperties { /** * Converts an array of host values to a comma-separated String. - * * It will append the default port value, if not already specified. + * + * @param hosts host string + * @param defaultPort port + * @return formatted connection string */ private String toConnectionString(String[] hosts, String defaultPort) { String[] fullyFormattedHosts = new String[hosts.length]; @@ -344,7 +350,7 @@ public class KafkaBinderConfigurationProperties { /** * No longer used. * @return the wait. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0") @@ -355,7 +361,7 @@ public class KafkaBinderConfigurationProperties { /** * No longer user. * @param maxWait the wait. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0") @@ -386,7 +392,7 @@ public class KafkaBinderConfigurationProperties { /** * No longer used. * @return the size. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0") @@ -397,7 +403,7 @@ public class KafkaBinderConfigurationProperties { /** * No longer used. * @param fetchSize the size. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0") @@ -424,7 +430,7 @@ public class KafkaBinderConfigurationProperties { /** * No longer used. * @return the queue size. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0") @@ -435,7 +441,7 @@ public class KafkaBinderConfigurationProperties { /** * No longer used. * @param queueSize the queue size. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0") @@ -463,7 +469,7 @@ public class KafkaBinderConfigurationProperties { * No longer used; set properties such as this via {@link #getConfiguration() * configuration}. * @return the size. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0, set properties such as this via 'configuration'") @@ -475,7 +481,7 @@ public class KafkaBinderConfigurationProperties { * No longer used; set properties such as this via {@link #getConfiguration() * configuration}. * @param socketBufferSize the size. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated @DeprecatedConfigurationProperty(reason = "Not used since 2.0, set properties such as this via 'configuration'") @@ -484,7 +490,7 @@ public class KafkaBinderConfigurationProperties { } public Map getConfiguration() { - return configuration; + return this.configuration; } public void setConfiguration(Map configuration) { @@ -540,7 +546,7 @@ public class KafkaBinderConfigurationProperties { Map producerConfiguration = new HashMap<>(); producerConfiguration.putAll(this.kafkaProperties.buildProducerProperties()); // Copy configured binder properties that apply to producers - for (Map.Entry configurationEntry : configuration.entrySet()) { + for (Map.Entry configurationEntry : this.configuration.entrySet()) { if (ProducerConfig.configNames().contains(configurationEntry.getKey())) { producerConfiguration.put(configurationEntry.getKey(), configurationEntry.getValue()); } @@ -585,6 +591,9 @@ public class KafkaBinderConfigurationProperties { this.headerMapperBeanName = headerMapperBeanName; } + /** + * Domain class that models transaction capabilities in Kafka. + */ public static class Transaction { private final CombinedProducerProperties producer = new CombinedProducerProperties(); diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBindingProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBindingProperties.java index 364f72bd6..121b4612b 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBindingProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBindingProperties.java @@ -19,17 +19,19 @@ package org.springframework.cloud.stream.binder.kafka.properties; import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider; /** + * Container object for Kafka specific extended producer and consumer binding properties. + * * @author Marius Bogoevici * @author Oleg Zhurakousky */ -public class KafkaBindingProperties implements BinderSpecificPropertiesProvider{ +public class KafkaBindingProperties implements BinderSpecificPropertiesProvider { private KafkaConsumerProperties consumer = new KafkaConsumerProperties(); private KafkaProducerProperties producer = new KafkaProducerProperties(); public KafkaConsumerProperties getConsumer() { - return consumer; + return this.consumer; } public void setConsumer(KafkaConsumerProperties consumer) { @@ -37,7 +39,7 @@ public class KafkaBindingProperties implements BinderSpecificPropertiesProvider{ } public KafkaProducerProperties getProducer() { - return producer; + return this.producer; } public void setProducer(KafkaProducerProperties producer) { diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java index 831c03131..a9b33e13c 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java @@ -21,6 +21,8 @@ import java.util.Map; /** + * Extended consumer properties for Kafka binder. + * * @author Marius Bogoevici * @author Ilayaperumal Gopinathan * @author Soby Chacko @@ -32,8 +34,17 @@ import java.util.Map; */ public class KafkaConsumerProperties { + /** + * Enumeration for starting consumer offset. + */ public enum StartOffset { + /** + * Starting from earliest offset. + */ earliest(-2L), + /** + * Starting from latest offset. + */ latest(-1L); private final long referencePoint; @@ -47,10 +58,25 @@ public class KafkaConsumerProperties { } } + /** + * Standard headers for the message. + */ public enum StandardHeaders { + /** + * No headers. + */ none, + /** + * Message header representing ID. + */ id, + /** + * Message header representing timestamp. + */ timestamp, + /** + * Indicating both ID and timestamp headers. + */ both } @@ -139,7 +165,7 @@ public class KafkaConsumerProperties { /** * No longer used. * @return the interval. - * @deprecated + * @deprecated No longer used by the binder */ @Deprecated public int getRecoveryInterval() { @@ -149,7 +175,7 @@ public class KafkaConsumerProperties { /** * No longer used. * @param recoveryInterval the interval. - * @deprecated + * @deprecated No longer needed by the binder */ @Deprecated public void setRecoveryInterval(int recoveryInterval) { @@ -173,7 +199,7 @@ public class KafkaConsumerProperties { } public String getDlqName() { - return dlqName; + return this.dlqName; } public void setDlqName(String dlqName) { @@ -181,7 +207,7 @@ public class KafkaConsumerProperties { } public String[] getTrustedPackages() { - return trustedPackages; + return this.trustedPackages; } public void setTrustedPackages(String[] trustedPackages) { @@ -189,7 +215,7 @@ public class KafkaConsumerProperties { } public KafkaProducerProperties getDlqProducerProperties() { - return dlqProducerProperties; + return this.dlqProducerProperties; } public void setDlqProducerProperties(KafkaProducerProperties dlqProducerProperties) { diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaExtendedBindingProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaExtendedBindingProperties.java index 5541f1f48..c1ef8daf8 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaExtendedBindingProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaExtendedBindingProperties.java @@ -21,6 +21,8 @@ import org.springframework.cloud.stream.binder.AbstractExtendedBindingProperties import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider; /** + * Kafka specific extended binding properties class that extends from {@link AbstractExtendedBindingProperties}. + * * @author Marius Bogoevici * @author Gary Russell * @author Soby Chacko diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java index 0dc4ce1cc..b722bc89c 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java @@ -24,6 +24,8 @@ import javax.validation.constraints.NotNull; import org.springframework.expression.Expression; /** + * Extended producer properties for Kafka binder. + * * @author Marius Bogoevici * @author Henryk Konsek * @author Gary Russell @@ -80,7 +82,7 @@ public class KafkaProducerProperties { } public Expression getMessageKeyExpression() { - return messageKeyExpression; + return this.messageKeyExpression; } public void setMessageKeyExpression(Expression messageKeyExpression) { @@ -111,10 +113,21 @@ public class KafkaProducerProperties { this.admin = admin; } - + /** + * Enumeration for compression types. + */ public enum CompressionType { + /** + * No compression. + */ none, + /** + * gzip based compression. + */ gzip, + /** + * snappy based compression. + */ snappy } } diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java index c7ed28f8d..3ac68c4b1 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java @@ -63,7 +63,7 @@ import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; /** - * Kafka implementation for {@link ProvisioningProvider} + * Kafka implementation for {@link ProvisioningProvider}. * * @author Soby Chacko * @author Gary Russell @@ -91,10 +91,12 @@ public class KafkaTopicProvisioner implements ProvisioningProvider> all = describeTopicsResult.all(); try { - Map topicDescriptions = all.get(operationTimeout, TimeUnit.SECONDS); + Map topicDescriptions = all.get(this.operationTimeout, TimeUnit.SECONDS); TopicDescription topicDescription = topicDescriptions.get(name); int partitions = topicDescription.partitions().size(); consumerDestination = createDlqIfNeedBe(adminClient, name, group, properties, anonymous, partitions); @@ -199,8 +201,8 @@ public class KafkaTopicProvisioner implements ProvisioningProvider> namesFutures = listTopicsResult.names(); - Set names = namesFutures.get(operationTimeout, TimeUnit.SECONDS); + Set names = namesFutures.get(this.operationTimeout, TimeUnit.SECONDS); if (names.contains(topicName)) { // only consider minPartitionCount for resizing if autoAddPartitions is true int effectivePartitionCount = this.configurationProperties.isAutoAddPartitions() @@ -316,17 +322,17 @@ public class KafkaTopicProvisioner implements ProvisioningProvider> topicDescriptionsFuture = describeTopicsResult.all(); - Map topicDescriptions = topicDescriptionsFuture.get(operationTimeout, TimeUnit.SECONDS); + Map topicDescriptions = topicDescriptionsFuture.get(this.operationTimeout, TimeUnit.SECONDS); TopicDescription topicDescription = topicDescriptions.get(topicName); int partitionSize = topicDescription.partitions().size(); if (partitionSize < effectivePartitionCount) { if (this.configurationProperties.isAutoAddPartitions()) { CreatePartitionsResult partitions = adminClient.createPartitions( Collections.singletonMap(topicName, NewPartitions.increaseTo(effectivePartitionCount))); - partitions.all().get(operationTimeout, TimeUnit.SECONDS); + partitions.all().get(this.operationTimeout, TimeUnit.SECONDS); } else if (tolerateLowerPartitionsOnBroker) { - logger.warn("The number of expected partitions was: " + partitionCount + ", but " + this.logger.warn("The number of expected partitions was: " + partitionCount + ", but " + partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead." + "There will be " + (effectivePartitionCount - partitionSize) + " idle consumers"); } @@ -342,7 +348,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider { + this.metadataRetryOperations.execute((context) -> { NewTopic newTopic; Map> replicasAssignments = adminProperties.getReplicasAssignments(); @@ -353,31 +359,31 @@ public class KafkaTopicProvisioner implements ProvisioningProvider 0) { newTopic.configs(adminProperties.getConfiguration()); } CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singletonList(newTopic)); try { - createTopicsResult.all().get(operationTimeout, TimeUnit.SECONDS); + createTopicsResult.all().get(this.operationTimeout, TimeUnit.SECONDS); } - catch (Exception e) { - if (e instanceof ExecutionException) { - String exceptionMessage = e.getMessage(); + catch (Exception ex) { + if (ex instanceof ExecutionException) { + String exceptionMessage = ex.getMessage(); if (exceptionMessage.contains("org.apache.kafka.common.errors.TopicExistsException")) { - if (logger.isWarnEnabled()) { - logger.warn("Attempt to create topic: " + topicName + ". Topic already exists."); + if (this.logger.isWarnEnabled()) { + this.logger.warn("Attempt to create topic: " + topicName + ". Topic already exists."); } } else { - logger.error("Failed to create topics", e.getCause()); - throw e.getCause(); + this.logger.error("Failed to create topics", ex.getCause()); + throw ex.getCause(); } } else { - logger.error("Failed to create topics", e.getCause()); - throw e.getCause(); + this.logger.error("Failed to create topics", ex.getCause()); + throw ex.getCause(); } } return null; @@ -390,13 +396,13 @@ public class KafkaTopicProvisioner implements ProvisioningProvider> callable) { try { return this.metadataRetryOperations - .execute(context -> { + .execute((context) -> { Collection partitions = callable.call(); // do a sanity check on the partition set int partitionSize = partitions.size(); if (partitionSize < partitionCount) { if (tolerateLowerPartitionsOnBroker) { - logger.warn("The number of expected partitions was: " + partitionCount + ", but " + this.logger.warn("The number of expected partitions was: " + partitionCount + ", but " + partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead." + "There will be " + (partitionCount - partitionSize) + " idle consumers"); } @@ -409,9 +415,9 @@ public class KafkaTopicProvisioner implements ProvisioningProvider