From 12f2084ae8a5b1f7e2f7bd63b79b82507b9d882b Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 8 Mar 2022 08:59:32 -0500 Subject: [PATCH] Switch to spring-kafka 3.0.0-SNAPSHOT - deprecations are temporarily restored there - fix deprecation warnings - re-enable tests for Windows --- build.gradle | 2 +- .../KafkaMessageListenerContainerSpec.java | 8 ++- .../kafka/inbound/KafkaInboundGateway.java | 3 +- .../KafkaMessageDrivenChannelAdapter.java | 3 +- .../kafka/channnel/ChannelTests.java | 5 +- .../kafka/config/xml/AllXmlTests.java | 5 +- ...essageDrivenChannelAdapterParserTests.java | 14 ++-- .../integration/kafka/dsl/KafkaDslTests.java | 6 +- .../kafka/inbound/InboundGatewayTests.java | 5 +- .../inbound/MessageDrivenAdapterTests.java | 9 +-- .../MessageSourceIntegrationTests.java | 5 +- .../kafka/inbound/MessageSourceTests.java | 64 +++++++++++++------ .../KafkaProducerMessageHandlerTests.java | 5 +- 13 files changed, 71 insertions(+), 63 deletions(-) diff --git a/build.gradle b/build.gradle index afe00003e9..fba9ccb9f7 100644 --- a/build.gradle +++ b/build.gradle @@ -100,7 +100,7 @@ ext { springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '3.0.0-SNAPSHOT' springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : '2022.0.0-SNAPSHOT' springGraphqlVersion = '1.0.0-SNAPSHOT' - springKafkaVersion = '3.0.0-M1' + springKafkaVersion = '3.0.0-SNAPSHOT' springRetryVersion = '1.3.1' springSecurityVersion = project.hasProperty('springSecurityVersion') ? project.springSecurityVersion : '6.0.0-SNAPSHOT' springVersion = project.hasProperty('springVersion') ? project.springVersion : '6.0.0-SNAPSHOT' diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java index f10a168e47..887c8e4130 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2021 the original author or authors. + * Copyright 2018-2022 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. @@ -26,7 +26,6 @@ import org.springframework.integration.dsl.IntegrationComponentSpec; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; import org.springframework.kafka.listener.ContainerProperties; -import org.springframework.kafka.listener.GenericErrorHandler; import org.springframework.kafka.support.TopicPartitionOffset; /** @@ -88,7 +87,10 @@ public class KafkaMessageListenerContainerSpec * @return the spec. * @see org.springframework.kafka.listener.ErrorHandler */ - public KafkaMessageListenerContainerSpec errorHandler(GenericErrorHandler errorHandler) { + @SuppressWarnings("deprecation") + public KafkaMessageListenerContainerSpec errorHandler( + org.springframework.kafka.listener.GenericErrorHandler errorHandler) { + this.target.setGenericErrorHandler(errorHandler); return this; } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java index a1edbb3793..3e0cbbbd34 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2021 the original author or authors. + * Copyright 2018-2022 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. @@ -170,6 +170,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem this.bindSourceRecord = bindSourceRecord; } + @SuppressWarnings("deprecation") @Override protected void onInit() { super.onInit(); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java index 03feaaef95..6e37423682 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2021 the original author or authors. + * Copyright 2015-2022 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. @@ -261,6 +261,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo return "kafka:message-driven-channel-adapter"; } + @SuppressWarnings("deprecation") @Override protected void onInit() { super.onInit(); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java index f5f7c21636..3972f94d5f 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2021 the original author or authors. + * Copyright 2020-2022 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. @@ -25,8 +25,6 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.condition.DisabledOnOs; -import org.junit.jupiter.api.condition.OS; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; @@ -59,7 +57,6 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; * @since 5.4 * */ -@DisabledOnOs(OS.WINDOWS) @SpringJUnitConfig @EmbeddedKafka(topics = { "channel.1", "channel.2", "channel.3" }, partitions = 1) public class ChannelTests { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java index d1e4108f23..c6866a6c5b 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2021 the original author or authors. + * Copyright 2019-2022 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. @@ -19,8 +19,6 @@ package org.springframework.integration.kafka.config.xml; import static org.assertj.core.api.Assertions.assertThat; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.condition.DisabledOnOs; -import org.junit.jupiter.api.condition.OS; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; @@ -36,7 +34,6 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; * @since 5.4 * */ -@DisabledOnOs(OS.WINDOWS) @SpringJUnitConfig @DirtiesContext @EmbeddedKafka(topics = { "one", "two", "three", "four" }) diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java index 1dcf78c25c..de125b4639 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2022 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. @@ -36,7 +36,6 @@ import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.KafkaMessageListenerContainer; import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter; import org.springframework.kafka.listener.adapter.RecordFilterStrategy; -import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter; import org.springframework.retry.RecoveryCallback; import org.springframework.retry.support.RetryTemplate; import org.springframework.test.annotation.DirtiesContext; @@ -110,7 +109,7 @@ class KafkaMessageDrivenChannelAdapterParserTests { } @Test - @SuppressWarnings("unchecked") + @SuppressWarnings({ "unchecked", "deprecation" }) void testKafkaMessageDrivenChannelAdapterOptions() { DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(Collections.emptyMap()); @@ -137,7 +136,8 @@ class KafkaMessageDrivenChannelAdapterParserTests { adapter.afterPropertiesSet(); messageListener = containerProps.getMessageListener(); - assertThat(messageListener).isInstanceOf(RetryingMessageListenerAdapter.class); + assertThat(messageListener).isInstanceOf( + org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.class); delegate = TestUtils.getPropertyValue(messageListener, "delegate"); @@ -151,13 +151,15 @@ class KafkaMessageDrivenChannelAdapterParserTests { delegate = TestUtils.getPropertyValue(messageListener, "delegate"); - assertThat(delegate).isInstanceOf(RetryingMessageListenerAdapter.class); + assertThat(delegate).isInstanceOf( + org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.class); adapter.setFilterInRetry(true); adapter.afterPropertiesSet(); messageListener = containerProps.getMessageListener(); - assertThat(messageListener).isInstanceOf(RetryingMessageListenerAdapter.class); + assertThat(messageListener).isInstanceOf( + org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.class); delegate = TestUtils.getPropertyValue(messageListener, "delegate"); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index 90f24f34c6..adeac2a2b0 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2021 the original author or authors. + * Copyright 2015-2022 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. @@ -32,8 +32,6 @@ import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.TopicPartition; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.condition.DisabledOnOs; -import org.junit.jupiter.api.condition.OS; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -95,7 +93,6 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; * * @since 5.4 */ -@DisabledOnOs(OS.WINDOWS) @SpringJUnitConfig @DirtiesContext @EmbeddedKafka(topics = { KafkaDslTests.TEST_TOPIC1, KafkaDslTests.TEST_TOPIC2, KafkaDslTests.TEST_TOPIC3, @@ -307,6 +304,7 @@ public class KafkaDslTests { .get(); } + @SuppressWarnings("deprecation") @Bean public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory factory = diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java index 2b1b02fe03..ab5b8e9374 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2021 the original author or authors. + * Copyright 2018-2022 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. @@ -32,8 +32,6 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.condition.DisabledOnOs; -import org.junit.jupiter.api.condition.OS; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.IntegrationMessageHeaderAccessor; @@ -78,7 +76,6 @@ import org.springframework.retry.support.RetryTemplate; * @since 5.4 * */ -@DisabledOnOs(OS.WINDOWS) class InboundGatewayTests { private static String topic1 = "testTopic1"; diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java index d579dd65cf..e5511b59db 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2022 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. @@ -52,8 +52,6 @@ import org.apache.kafka.common.header.internals.RecordHeaders; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.condition.DisabledOnOs; -import org.junit.jupiter.api.condition.OS; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.StaticMessageHeaderAccessor; @@ -73,8 +71,8 @@ import org.springframework.kafka.event.ConsumerPausedEvent; import org.springframework.kafka.event.ConsumerResumedEvent; import org.springframework.kafka.event.KafkaEvent; import org.springframework.kafka.listener.ContainerProperties; +import org.springframework.kafka.listener.DefaultErrorHandler; import org.springframework.kafka.listener.KafkaMessageListenerContainer; -import org.springframework.kafka.listener.SeekToCurrentErrorHandler; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.DefaultKafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; @@ -111,7 +109,6 @@ import org.springframework.retry.support.RetryTemplate; * @since 5.4 * */ -@DisabledOnOs(OS.WINDOWS) class MessageDrivenAdapterTests { private static String topic1 = "testTopic1"; @@ -343,7 +340,7 @@ class MessageDrivenAdapterTests { containerProps.setDeliveryAttemptHeader(true); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); - container.setErrorHandler(new SeekToCurrentErrorHandler()); + container.setCommonErrorHandler(new DefaultErrorHandler()); KafkaMessageDrivenChannelAdapter adapter = new KafkaMessageDrivenChannelAdapter<>(container); MessageChannel out = new DirectChannel() { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java index 41fb21c240..1a11c39e07 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2021 the original author or authors. + * Copyright 2018-2022 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. @@ -29,8 +29,6 @@ import org.apache.kafka.common.TopicPartition; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.condition.DisabledOnOs; -import org.junit.jupiter.api.condition.OS; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; @@ -48,7 +46,6 @@ import org.springframework.messaging.Message; * @since 5.4 * */ -@DisabledOnOs(OS.WINDOWS) class MessageSourceIntegrationTests { private static final String TOPIC1 = "MessageSourceIntegrationTests1"; diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java index bbc7b1f7cd..becd5e701c 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2020 the original author or authors. + * Copyright 2018-2022 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. @@ -43,6 +43,7 @@ import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -63,6 +64,7 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.consumer.OffsetCommitCallback; import org.apache.kafka.clients.consumer.OffsetResetStrategy; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.record.TimestampType; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; @@ -251,16 +253,20 @@ class MessageSourceTests { }).given(consumer).commitAsync(any(), any()); Map> records1 = new LinkedHashMap<>(); records1.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "foo", + new RecordHeaders(), Optional.empty()))); Map> records2 = new LinkedHashMap<>(); records2.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "bar", + new RecordHeaders(), Optional.empty()))); Map> records3 = new LinkedHashMap<>(); records3.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"))); + new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "baz", + new RecordHeaders(), Optional.empty()))); Map> records4 = new LinkedHashMap<>(); records4.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 3L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + new ConsumerRecord("foo", 0, 3L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "qux", + new RecordHeaders(), Optional.empty()))); ConsumerRecords cr1 = new ConsumerRecords(records1); ConsumerRecords cr2 = new ConsumerRecords(records2); ConsumerRecords cr3 = new ConsumerRecords(records3); @@ -375,22 +381,28 @@ class MessageSourceTests { willAnswer(i -> paused.get()).given(consumer).paused(); Map> records1 = new LinkedHashMap<>(); records1.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "foo", + new RecordHeaders(), Optional.empty()))); Map> records2 = new LinkedHashMap<>(); records2.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "bar", + new RecordHeaders(), Optional.empty()))); Map> records3 = new LinkedHashMap<>(); records3.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"))); + new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "baz", + new RecordHeaders(), Optional.empty()))); Map> records4 = new LinkedHashMap<>(); records4.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 3L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + new ConsumerRecord("foo", 0, 3L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "qux", + new RecordHeaders(), Optional.empty()))); Map> records5 = new LinkedHashMap<>(); records5.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 4L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "fiz"))); + new ConsumerRecord("foo", 0, 4L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "fiz", + new RecordHeaders(), Optional.empty()))); Map> records6 = new LinkedHashMap<>(); records6.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 5L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "buz"))); + new ConsumerRecord("foo", 0, 5L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "buz", + new RecordHeaders(), Optional.empty()))); ConsumerRecords cr1 = new ConsumerRecords(records1); ConsumerRecords cr2 = new ConsumerRecords(records2); ConsumerRecords cr3 = new ConsumerRecords(records3); @@ -465,11 +477,13 @@ class MessageSourceTests { willAnswer(i -> paused.get()).given(consumer).paused(); Map> records1 = new LinkedHashMap<>(); records1.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "foo", + new RecordHeaders(), Optional.empty()))); ConsumerRecords cr1 = new ConsumerRecords(records1); Map> records2 = new LinkedHashMap<>(); records2.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "bar", + new RecordHeaders(), Optional.empty()))); ConsumerRecords cr2 = new ConsumerRecords(records2); ConsumerRecords cr3 = new ConsumerRecords(Collections.emptyMap()); given(consumer.poll(any(Duration.class))).willReturn(cr1, cr1, cr2, cr2, cr3); @@ -534,11 +548,13 @@ class MessageSourceTests { willAnswer(i -> paused.get()).given(consumer).paused(); Map> records1 = new LinkedHashMap<>(); records1.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "foo", + new RecordHeaders(), Optional.empty()))); ConsumerRecords cr1 = new ConsumerRecords(records1); Map> records2 = new LinkedHashMap<>(); records2.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"))); + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "bar", + new RecordHeaders(), Optional.empty()))); ConsumerRecords cr2 = new ConsumerRecords(records2); ConsumerRecords cr3 = new ConsumerRecords(Collections.emptyMap()); given(consumer.poll(any(Duration.class))).willReturn(cr1, cr2, cr1, cr2, cr3); @@ -635,12 +651,14 @@ class MessageSourceTests { Map> records1 = new LinkedHashMap<>(); records1.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "foo", + new RecordHeaders(), Optional.empty()))); ConsumerRecords cr1 = new ConsumerRecords(records1); given(consumer.poll(Duration.of(20 * 5000, ChronoUnit.MILLIS))).willReturn(cr1, ConsumerRecords.EMPTY); Map> records2 = new LinkedHashMap<>(); records2.put(topicPartition, Collections.singletonList( - new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "foo", + new RecordHeaders(), Optional.empty()))); ConsumerRecords cr2 = new ConsumerRecords(records2); given(consumer.poll(Duration.of(5000, ChronoUnit.MILLIS))).willReturn(cr2, ConsumerRecords.EMPTY); ConsumerFactory consumerFactory = mock(ConsumerFactory.class); @@ -698,10 +716,14 @@ class MessageSourceTests { willDoNothing().given(consumer).resume(partitions.capture()); Map> records = new LinkedHashMap<>(); records.put(topicPartition, Arrays.asList( - new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"), - new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar"), - new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz"), - new ConsumerRecord("foo", 0, 3L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "qux"))); + new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "foo", + new RecordHeaders(), Optional.empty()), + new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "bar", + new RecordHeaders(), Optional.empty()), + new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "baz", + new RecordHeaders(), Optional.empty()), + new ConsumerRecord("foo", 0, 3L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "qux", + new RecordHeaders(), Optional.empty()))); ConsumerRecords cr1 = new ConsumerRecords(records); ConsumerRecords cr2 = new ConsumerRecords(Collections.emptyMap()); given(consumer.poll(any(Duration.class))).willReturn(cr1, cr2); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index e9e2a8fbc8..da21723ffb 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2022 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. @@ -62,8 +62,6 @@ import org.apache.kafka.common.TopicPartition; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.condition.DisabledOnOs; -import org.junit.jupiter.api.condition.OS; import org.mockito.ArgumentCaptor; import org.mockito.InOrder; @@ -120,7 +118,6 @@ import org.springframework.util.concurrent.SettableListenableFuture; * * @since 5.4 */ -@DisabledOnOs(OS.WINDOWS) class KafkaProducerMessageHandlerTests { private static String topic1 = "testTopic1out";