From 9d212024f811204158be1c7fec4559857093e496 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 4 Mar 2020 18:28:42 -0500 Subject: [PATCH] Updates for 3.1.0 spring-cloud-build to 3.0.0 snapshot spring kafka to 2.4.x snapshot SIK 3.2.1 Remove a test that has behavior inconsistent with new changes in Spring Kafka 2.4 where all error handlers have isAckAfterHandle() true by default. The test for auto commit offset on error without dlq was expecting this acknowledgement to not to occur. If applications need to have the ack turned off on error, they should provide a container customizer where it sets the ack to false. Since this is not a binder concern, we are removing the test testDefaultAutoCommitOnErrorWithoutDlq. Cleaning up in Kafka Streams binder tests. --- pom.xml | 6 +- .../MultipleFunctionsInSameAppTests.java | 3 + .../stream/binder/kafka/KafkaBinderTests.java | 71 +------------------ 3 files changed, 7 insertions(+), 73 deletions(-) diff --git a/pom.xml b/pom.xml index e575534e5..968e39746 100644 --- a/pom.xml +++ b/pom.xml @@ -7,14 +7,14 @@ org.springframework.cloud spring-cloud-build - 2.2.2.RELEASE + 3.0.0.BUILD-SNAPSHOT 1.8 - 2.3.5.RELEASE + 2.4.4.BUILD-SNAPSHOT 3.2.1.RELEASE - 2.3.1 + 2.4.0 1.1.0.BUILD-SNAPSHOT 3.1.0.BUILD-SNAPSHOT true diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java index 8a1af8ed3..445072bc9 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java @@ -131,6 +131,7 @@ public class MultipleFunctionsInSameAppTests { "--spring.jmx.enabled=false", "--spring.cloud.stream.function.definition=process;analyze", "--spring.cloud.stream.bindings.process-in-0.destination=purchases", + "--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.startOffset=latest", "--spring.cloud.stream.bindings.process-in-0.binder=kafka1", "--spring.cloud.stream.bindings.process-out-0.destination=coffee", "--spring.cloud.stream.bindings.process-out-0.binder=kafka1", @@ -148,6 +149,8 @@ public class MultipleFunctionsInSameAppTests { "--spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.applicationId=my-app-2", "--spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.configuration.client.id=analyze-client")) { + + Thread.sleep(1000); receiveAndValidate("purchases", "coffee", "electronics"); StreamsBuilderFactoryBean processStreamsBuilderFactoryBean = context diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 9f0cf8a3a..fdb8dfdfc 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -1108,75 +1108,6 @@ public class KafkaBinderTests extends producerBinding.unbind(); } - @Test - @SuppressWarnings("unchecked") - public void testDefaultAutoCommitOnErrorWithoutDlq() throws Exception { - Binder binder = getBinder(); - - ExtendedProducerProperties producerProperties = createProducerProperties(); - BindingProperties producerBindingProperties = createProducerBindingProperties( - producerProperties); - - DirectChannel moduleOutputChannel = createBindableChannel("output", - producerBindingProperties); - - ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - consumerProperties.setMaxAttempts(1); - consumerProperties.setBackOffInitialInterval(100); - consumerProperties.setBackOffMaxInterval(150); - consumerProperties.getExtension().setAutoRebalanceEnabled(false); - - DirectChannel moduleInputChannel = createBindableChannel("input", - createConsumerBindingProperties(consumerProperties)); - - FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler(); - moduleInputChannel.subscribe(handler); - - long uniqueBindingId = System.currentTimeMillis(); - Binding producerBinding = binder.bindProducer( - "retryTest." + uniqueBindingId + ".0", moduleOutputChannel, - producerProperties); - Binding consumerBinding = binder.bindConsumer( - "retryTest." + uniqueBindingId + ".0", "testGroup", moduleInputChannel, - consumerProperties); - - String testMessagePayload = "test." + UUID.randomUUID().toString(); - Message testMessage = MessageBuilder - .withPayload(testMessagePayload.getBytes()).build(); - moduleOutputChannel.send(testMessage); - - assertThat(handler.getLatch().await((int) (timeoutMultiplier * 1000), - TimeUnit.MILLISECONDS)); - // first attempt fails - assertThat(handler.getReceivedMessages().entrySet()).hasSize(1); - Message receivedMessage = handler.getReceivedMessages().entrySet().iterator() - .next().getValue(); - assertThat(receivedMessage).isNotNull(); - assertThat( - new String((byte[]) receivedMessage.getPayload(), StandardCharsets.UTF_8)) - .isEqualTo(testMessagePayload); - assertThat(handler.getInvocationCount()) - .isEqualTo(consumerProperties.getMaxAttempts()); - consumerBinding.unbind(); - - // on the second attempt the message is redelivered - QueueChannel successfulInputChannel = new QueueChannel(); - consumerBinding = binder.bindConsumer("retryTest." + uniqueBindingId + ".0", - "testGroup", successfulInputChannel, consumerProperties); - binderBindUnbindLatency(); - String testMessage2Payload = "test." + UUID.randomUUID().toString(); - Message testMessage2 = MessageBuilder - .withPayload(testMessage2Payload.getBytes()).build(); - moduleOutputChannel.send(testMessage2); - - Message firstReceived = receive(successfulInputChannel); - assertThat(firstReceived.getPayload()).isEqualTo(testMessagePayload.getBytes()); - Message secondReceived = receive(successfulInputChannel); - assertThat(secondReceived.getPayload()).isEqualTo(testMessage2Payload.getBytes()); - consumerBinding.unbind(); - producerBinding.unbind(); - } - @Test @SuppressWarnings("unchecked") public void testDefaultAutoCommitOnErrorWithDlq() throws Exception { @@ -1730,7 +1661,7 @@ public class KafkaBinderTests extends } catch (UnsupportedOperationException ignored) { } - List interceptors = output.getChannelInterceptors(); + List interceptors = output.getInterceptors(); AtomicInteger count = new AtomicInteger(); interceptors.forEach(interceptor -> { if (interceptor instanceof PartitioningInterceptor) {