Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Switch to spring-kafka 3.0.0-SNAPSHOT

Browse Source 
- deprecations are temporarily restored there
 - fix deprecation warnings
 - re-enable tests for Windows
This commit is contained in:
Gary Russell
2022-03-08 08:59:32 -05:00
committed by Artem Bilan
parent 70587f5e2f
commit 12f2084ae8
13 changed files with 71 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -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'

8
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java
View File

@@ -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<K, V>
* @return the spec.
* @see org.springframework.kafka.listener.ErrorHandler
*/
public KafkaMessageListenerContainerSpec<K, V> errorHandler(GenericErrorHandler<?> errorHandler) {
@SuppressWarnings("deprecation")
public KafkaMessageListenerContainerSpec<K, V> errorHandler(
org.springframework.kafka.listener.GenericErrorHandler<?> errorHandler) {
this.target.setGenericErrorHandler(errorHandler);
return this;
}

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java
View File

@@ -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<K, V, R> extends MessagingGatewaySupport implem
this.bindSourceRecord = bindSourceRecord;
}
@SuppressWarnings("deprecation")
@Override
protected void onInit() {
super.onInit();

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java
View File

@@ -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<K, V> extends MessageProducerSuppo
return "kafka:message-driven-channel-adapter";
}
@SuppressWarnings("deprecation")
@Override
protected void onInit() {
super.onInit();

5
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java
View File

@@ -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 {

5
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java
View File

@@ -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" })

14
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java
View File

@@ -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<Integer, String> 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");

6
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -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<Integer, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =

5
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java
View File

@@ -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";

9
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
View File

@@ -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<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
container.setErrorHandler(new SeekToCurrentErrorHandler());
container.setCommonErrorHandler(new DefaultErrorHandler());
KafkaMessageDrivenChannelAdapter<Integer, String> adapter = new KafkaMessageDrivenChannelAdapter<>(container);
MessageChannel out = new DirectChannel() {

5
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java
View File

@@ -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";

64
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java
View File

@@ -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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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<TopicPartition, List<ConsumerRecord>> 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);

5
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
View File

@@ -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";