From f596e49242045467830a499b4ac08fd16482e00d Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 14 Feb 2020 17:57:32 -0500 Subject: [PATCH] Upgrade to latest releases * Upgrade to Gradle 6.1.1 * Fix for latest Checkstyle * Fix for latest SI-Kotlin-DSL --- .../outbound/KafkaProducerMessageHandler.java | 4 +- .../dsl/{ => kotlin}/KafkaDslKotlinTests.kt | 51 +++++++++---------- 2 files changed, 27 insertions(+), 28 deletions(-) rename spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/{ => kotlin}/KafkaDslKotlinTests.kt (91%) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index ae616401c9..f7a6cd41a2 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2019 the original author or authors. + * Copyright 2013-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -522,7 +522,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes } private void determineValidReplyTopicsAndPartitions() { - ReplyingKafkaTemplate rkt = (ReplyingKafkaTemplate) kafkaTemplate; + ReplyingKafkaTemplate rkt = (ReplyingKafkaTemplate) this.kafkaTemplate; Collection replyTopics = rkt.getAssignedReplyTopicPartitions(); Map> topicsAndPartitions = new HashMap<>(); if (replyTopics != null) { diff --git a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt similarity index 91% rename from spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt rename to spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt index c68d1b0d36..f232047011 100644 --- a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt +++ b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.kafka.dsl +package org.springframework.integration.kafka.dsl.kotlin import assertk.assertThat import assertk.assertions.isEqualTo @@ -27,7 +27,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig import org.apache.kafka.clients.consumer.ConsumerRebalanceListener import org.apache.kafka.clients.producer.ProducerConfig import org.apache.kafka.common.TopicPartition -import org.junit.Test +import org.junit.jupiter.api.Test import org.springframework.beans.factory.annotation.Autowired import org.springframework.beans.factory.annotation.Qualifier import org.springframework.context.annotation.Bean @@ -37,8 +37,9 @@ import org.springframework.integration.MessageRejectedException import org.springframework.integration.channel.QueueChannel import org.springframework.integration.config.EnableIntegration import org.springframework.integration.dsl.Pollers -import org.springframework.integration.dsl.integrationFlow +import org.springframework.integration.dsl.kotlin.integrationFlow import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer +import org.springframework.integration.kafka.dsl.Kafka import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy @@ -251,9 +252,9 @@ class KafkaDslKotlinTests { RawRecordHeaderErrorMessageStrategy())) .retryTemplate(RetryTemplate()) .filterInRetry(true)) { - filter>({ m -> (m.headers[KafkaHeaders.RECEIVED_MESSAGE_KEY] as Int) < 101 }) { f -> f.throwExceptionOnRejection(true) } - transform({ it.toUpperCase() }) - channel { c -> c.queue("listeningFromKafkaResults1") } + filter>({ m -> (m.headers[KafkaHeaders.RECEIVED_MESSAGE_KEY] as Int) < 101 }) { throwExceptionOnRejection(true) } + transform { it.toUpperCase() } + channel { queue("listeningFromKafkaResults1") } } @Bean @@ -266,9 +267,9 @@ class KafkaDslKotlinTests { RawRecordHeaderErrorMessageStrategy())) .retryTemplate(RetryTemplate()) .filterInRetry(true)) { - filter>({ m -> (m.headers[KafkaHeaders.RECEIVED_MESSAGE_KEY] as Int) < 101 }) { it.throwExceptionOnRejection(true) } - transform({ it.toUpperCase() }) - channel { c -> c.queue("listeningFromKafkaResults2") } + filter>({ m -> (m.headers[KafkaHeaders.RECEIVED_MESSAGE_KEY] as Int) < 101 }) { throwExceptionOnRejection(true) } + transform { it.toUpperCase() } + channel { queue("listeningFromKafkaResults2") } } @Bean @@ -281,20 +282,18 @@ class KafkaDslKotlinTests { @Bean fun sendToKafkaFlow() = integrationFlow { - split({ p -> Stream.generate { p }.limit(101) }) + split { p -> Stream.generate { p }.limit(101) } publishSubscribeChannel { - it - .subscribe { - it.handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC1) - .timestampExpression("T(Long).valueOf('1487694048633')") - ) { it.id("kafkaProducer1") } - } - .subscribe { - it.handle( - kafkaMessageHandler(producerFactory(), TEST_TOPIC2) - .timestamp { 1487694048644L } - ) { it.id("kafkaProducer2") } - } + subscribe(integrationFlow { + handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC1) + .timestampExpression("T(Long).valueOf('1487694048633')") + ) { id("kafkaProducer1") } + }) + subscribe(integrationFlow { + handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC2) + .timestamp { 1487694048644L } + ) { id("kafkaProducer2") } + }) } } @@ -303,7 +302,7 @@ class KafkaDslKotlinTests { private fun kafkaMessageHandler(producerFactory: ProducerFactory, topic: String) = Kafka.outboundChannelAdapter(producerFactory) - .messageKey { m -> m.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] } + .messageKey { it.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] } .headerMapper(mapper()) .sync(true) .partitionId { 0 } @@ -314,7 +313,7 @@ class KafkaDslKotlinTests { @Bean fun sourceFlow() = integrationFlow(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3)), - { e -> e.poller(Pollers.fixedDelay(100)) }) { + { poller(Pollers.fixedDelay(100)) }) { handle { m -> this@ContextConfiguration.fromSource = m.payload this@ContextConfiguration.sourceFlowLatch.countDown() @@ -355,7 +354,7 @@ class KafkaDslKotlinTests { @Bean fun serverGateway() = integrationFlow(Kafka.inboundGateway(consumerFactory(), containerProperties(), producerFactory())) { - transform({ it.toUpperCase() }) + transform { it.toUpperCase() } } private fun containerProperties() =