Skip to content

Commit 47a728f

Browse files
ChaedieChaedie
committed
refactor: replace deprecated code
Signed-off-by: Chaedong Im <[email protected]>
1 parent df86512 commit 47a728f

File tree

2 files changed

+45
-32
lines changed

2 files changed

+45
-32
lines changed

‎spring-kafka/src/test/java/org/springframework/kafka/listener/AckModeRecordFilteredTest.java‎

Lines changed: 23 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -16,24 +16,31 @@
1616

1717
package org.springframework.kafka.listener;
1818

19+
import java.time.Duration;
20+
import java.util.ArrayList;
21+
import java.util.HashMap;
22+
import java.util.List;
23+
import java.util.Map;
24+
import java.util.concurrent.CountDownLatch;
25+
import java.util.concurrent.TimeUnit;
26+
1927
import org.apache.kafka.clients.consumer.Consumer;
2028
import org.apache.kafka.clients.consumer.ConsumerRecord;
2129
import org.apache.kafka.clients.consumer.ConsumerRecords;
2230
import org.apache.kafka.common.TopicPartition;
2331
import org.junit.jupiter.api.Test;
32+
2433
import org.springframework.kafka.core.ConsumerFactory;
2534
import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
2635
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
2736

28-
import java.time.Duration;
29-
import java.util.*;
30-
import java.util.concurrent.CountDownLatch;
31-
import java.util.concurrent.TimeUnit;
32-
3337
import static org.assertj.core.api.Assertions.assertThat;
3438
import static org.mockito.ArgumentMatchers.any;
3539
import static org.mockito.BDDMockito.given;
36-
import static org.mockito.Mockito.*;
40+
import static org.mockito.Mockito.mock;
41+
import static org.mockito.Mockito.never;
42+
import static org.mockito.Mockito.times;
43+
import static org.mockito.Mockito.verify;
3744

3845
/**
3946
* Tests for the RECORD_FILTERED acknowledge mode.
@@ -45,7 +52,7 @@
4552
*/
4653
public class AckModeRecordFilteredTest {
4754

48-
@SuppressWarnings("unchecked")
55+
@SuppressWarnings({"unchecked", "deprecation"})
4956
@Test
5057
public void testRecordFilteredModeOnlyCommitsProcessedRecords() throws InterruptedException {
5158
// Given: A container with RECORD_FILTERED ack mode
@@ -88,7 +95,7 @@ public void testRecordFilteredModeOnlyCommitsProcessedRecords() throws Interrupt
8895

8996
given(consumer.poll(any(Duration.class)))
9097
.willReturn(consumerRecords)
91-
.willReturn(new ConsumerRecords<>(Collections.emptyMap()));
98+
.willReturn(ConsumerRecords.empty());
9299

93100
// When: Start the container and process records
94101
container.start();
@@ -102,7 +109,7 @@ public void testRecordFilteredModeOnlyCommitsProcessedRecords() throws Interrupt
102109
verify(consumer, times(2)).commitSync(any(), any(Duration.class));
103110
}
104111

105-
@SuppressWarnings("unchecked")
112+
@SuppressWarnings({"unchecked", "deprecation"})
106113
@Test
107114
public void testRecordFilteredModeWithAllRecordsFiltered() throws InterruptedException {
108115
// Given: All records are filtered
@@ -139,7 +146,7 @@ public void testRecordFilteredModeWithAllRecordsFiltered() throws InterruptedExc
139146

140147
given(consumer.poll(any(Duration.class)))
141148
.willReturn(consumerRecords)
142-
.willReturn(new ConsumerRecords<>(Collections.emptyMap()));
149+
.willReturn(ConsumerRecords.empty());
143150

144151
// When
145152
container.start();
@@ -150,7 +157,7 @@ public void testRecordFilteredModeWithAllRecordsFiltered() throws InterruptedExc
150157
verify(consumer, never()).commitSync(any(), any(Duration.class));
151158
}
152159

153-
@SuppressWarnings("unchecked")
160+
@SuppressWarnings({"unchecked", "deprecation"})
154161
@Test
155162
public void testRecordFilteredModeWithMixedPartitions() throws InterruptedException {
156163
// Given: Mixed partitions with different filtering scenarios
@@ -201,7 +208,7 @@ public void testRecordFilteredModeWithMixedPartitions() throws InterruptedExcept
201208

202209
given(consumer.poll(any(Duration.class)))
203210
.willReturn(consumerRecords)
204-
.willReturn(new ConsumerRecords<>(Collections.emptyMap()));
211+
.willReturn(ConsumerRecords.empty());
205212

206213
// When
207214
container.start();
@@ -213,7 +220,7 @@ public void testRecordFilteredModeWithMixedPartitions() throws InterruptedExcept
213220
verify(consumer, times(3)).commitSync(any(), any(Duration.class));
214221
}
215222

216-
@SuppressWarnings("unchecked")
223+
@SuppressWarnings({"unchecked", "deprecation"})
217224
@Test
218225
public void testRecordFilteredModeEfficiencyGains() throws InterruptedException {
219226
ConsumerFactory<String, String> consumerFactory = mock(ConsumerFactory.class);
@@ -253,7 +260,7 @@ public void testRecordFilteredModeEfficiencyGains() throws InterruptedException
253260

254261
given(consumer.poll(any(Duration.class)))
255262
.willReturn(consumerRecords)
256-
.willReturn(new ConsumerRecords<>(Collections.emptyMap()));
263+
.willReturn(ConsumerRecords.empty());
257264

258265
// When
259266
container.start();
@@ -266,7 +273,7 @@ public void testRecordFilteredModeEfficiencyGains() throws InterruptedException
266273
verify(consumer, times(1)).commitSync(any(), any(Duration.class));
267274
}
268275

269-
@SuppressWarnings("unchecked")
276+
@SuppressWarnings({"unchecked", "deprecation"})
270277
@Test
271278
public void testRecordFilteredModeDoesNotBreakNormalProcessing() throws InterruptedException {
272279
ConsumerFactory<String, String> consumerFactory = mock(ConsumerFactory.class);
@@ -307,7 +314,7 @@ public void testRecordFilteredModeDoesNotBreakNormalProcessing() throws Interrup
307314

308315
given(consumer.poll(any(Duration.class)))
309316
.willReturn(consumerRecords)
310-
.willReturn(new ConsumerRecords<>(Collections.emptyMap()));
317+
.willReturn(ConsumerRecords.empty());
311318

312319
// When
313320
container.start();

‎spring-kafka/src/test/java/org/springframework/kafka/listener/AckModeRecordWithFilteringTest.java‎

Lines changed: 22 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -16,25 +16,31 @@
1616

1717
package org.springframework.kafka.listener;
1818

19+
import java.time.Duration;
20+
import java.util.ArrayList;
21+
import java.util.HashMap;
22+
import java.util.List;
23+
import java.util.Map;
24+
import java.util.concurrent.CountDownLatch;
25+
import java.util.concurrent.TimeUnit;
26+
1927
import org.apache.kafka.clients.consumer.Consumer;
2028
import org.apache.kafka.clients.consumer.ConsumerRecord;
2129
import org.apache.kafka.clients.consumer.ConsumerRecords;
2230
import org.apache.kafka.common.TopicPartition;
2331
import org.junit.jupiter.api.Test;
32+
2433
import org.springframework.kafka.core.ConsumerFactory;
2534
import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
2635
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
2736

28-
import java.time.Duration;
29-
import java.util.*;
30-
import java.util.concurrent.CountDownLatch;
31-
import java.util.concurrent.TimeUnit;
32-
3337
import static org.assertj.core.api.Assertions.assertThat;
3438
import static org.mockito.ArgumentMatchers.any;
3539
import static org.mockito.ArgumentMatchers.anyMap;
3640
import static org.mockito.BDDMockito.given;
37-
import static org.mockito.Mockito.*;
41+
import static org.mockito.Mockito.mock;
42+
import static org.mockito.Mockito.times;
43+
import static org.mockito.Mockito.verify;
3844

3945
/**
4046
* Tests to verify the behavior of RECORD acknowledge mode when used with filtering strategies.
@@ -46,7 +52,7 @@
4652
*/
4753
public class AckModeRecordWithFilteringTest {
4854

49-
@SuppressWarnings("unchecked")
55+
@SuppressWarnings({"unchecked", "deprecation"})
5056
@Test
5157
public void testCurrentRecordModeCommitsAllRecords() throws InterruptedException {
5258
// Given: A container with RECORD ack mode and a filter that filters out even offsets
@@ -89,7 +95,7 @@ public void testCurrentRecordModeCommitsAllRecords() throws InterruptedException
8995

9096
given(consumer.poll(any(Duration.class)))
9197
.willReturn(consumerRecords)
92-
.willReturn(new ConsumerRecords<>(Collections.emptyMap()));
98+
.willReturn(ConsumerRecords.empty());
9399

94100
// When: Start the container and process records
95101
container.start();
@@ -103,7 +109,7 @@ public void testCurrentRecordModeCommitsAllRecords() throws InterruptedException
103109
verify(consumer, times(4)).commitSync(any(), any(Duration.class));
104110
}
105111

106-
@SuppressWarnings("unchecked")
112+
@SuppressWarnings({"unchecked", "deprecation"})
107113
@Test
108114
public void testAllRecordsFilteredStillCommits() throws InterruptedException {
109115
// Given: A container where all records are filtered
@@ -139,7 +145,7 @@ public void testAllRecordsFilteredStillCommits() throws InterruptedException {
139145

140146
given(consumer.poll(any(Duration.class)))
141147
.willReturn(consumerRecords)
142-
.willReturn(new ConsumerRecords<>(Collections.emptyMap()));
148+
.willReturn(ConsumerRecords.empty());
143149

144150
// When: Start the container
145151
container.start();
@@ -151,7 +157,7 @@ public void testAllRecordsFilteredStillCommits() throws InterruptedException {
151157
verify(consumer, times(2)).commitSync(any(), any(Duration.class));
152158
}
153159

154-
@SuppressWarnings("unchecked")
160+
@SuppressWarnings({"unchecked", "deprecation"})
155161
@Test
156162
public void testMixedPartitionsWithFiltering() throws InterruptedException {
157163
// Given: Multiple partitions with different records
@@ -201,7 +207,7 @@ record -> record.value().contains("skip");
201207

202208
given(consumer.poll(any(Duration.class)))
203209
.willReturn(consumerRecords)
204-
.willReturn(new ConsumerRecords<>(Collections.emptyMap()));
210+
.willReturn(ConsumerRecords.empty());
205211

206212
// When: Start container
207213
container.start();
@@ -215,7 +221,7 @@ record -> record.value().contains("skip");
215221
verify(consumer, times(5)).commitSync(any(), any(Duration.class));
216222
}
217223

218-
@SuppressWarnings("unchecked")
224+
@SuppressWarnings({"unchecked", "deprecation"})
219225
@Test
220226
public void testCommitLogging() throws InterruptedException {
221227
ConsumerFactory<String, String> consumerFactory = mock(ConsumerFactory.class);
@@ -251,7 +257,7 @@ public void testCommitLogging() throws InterruptedException {
251257

252258
given(consumer.poll(any(Duration.class)))
253259
.willReturn(consumerRecords)
254-
.willReturn(new ConsumerRecords<>(Collections.emptyMap()));
260+
.willReturn(ConsumerRecords.empty());
255261

256262
// When
257263
container.start();
@@ -262,7 +268,7 @@ public void testCommitLogging() throws InterruptedException {
262268
verify(consumer, times(2)).commitSync(anyMap(), any(Duration.class));
263269
}
264270

265-
@SuppressWarnings("unchecked")
271+
@SuppressWarnings({"unchecked", "deprecation"})
266272
@Test
267273
public void testAckDiscardedParameterBehavior() throws InterruptedException {
268274
ConsumerFactory<String, String> consumerFactory = mock(ConsumerFactory.class);
@@ -303,7 +309,7 @@ public void testAckDiscardedParameterBehavior() throws InterruptedException {
303309

304310
given(consumer.poll(any(Duration.class)))
305311
.willReturn(consumerRecords)
306-
.willReturn(new ConsumerRecords<>(Collections.emptyMap()));
312+
.willReturn(ConsumerRecords.empty());
307313

308314
container.start();
309315
assertThat(processedLatch.await(5, TimeUnit.SECONDS)).isTrue();

0 commit comments

Comments
 (0)