You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Copilot (via GitHub)" <gi...@apache.org> on 2025/12/01 13:02:04 UTC
Re: [PR] KAFKA-19942: Clean up StreamThread and StoreChangelogReader Test [kafka]
Copilot commented on code in PR #21022:
URL: https://github.com/apache/kafka/pull/21022#discussion_r2576990126
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java:
##########
@@ -2548,26 +2311,18 @@ public Set<TopicPartition> partitions() {
"K2".getBytes(),
"V2".getBytes()));
- if (stateUpdaterEnabled) {
- TestUtils.waitForCondition(
- () -> mockRestoreConsumer.assignment().isEmpty(),
- "Never get the assignment");
- } else {
- TestUtils.waitForCondition(
- () -> {
- mockRestoreConsumer.assign(changelogPartitionSet);
- return mockRestoreConsumer.position(changelogPartition) == 2L;
- },
- "Never finished restore");
- }
+ TestUtils.waitForCondition(
+ () -> mockRestoreConsumer.assignment().isEmpty(),
+ "Never get the assignment");
+
Review Comment:
Trailing whitespace detected. Please remove the trailing whitespace on this line for consistency with the project's code style.
```suggestion
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java:
##########
@@ -1768,15 +1724,14 @@ public void shouldReinitializeRevivedTasksInAnyState(final boolean stateUpdaterE
runOnce(processingThreadsEnabled);
// the third actually polls, processes the record, and throws the corruption exception
- if (stateUpdaterEnabled) {
- TestUtils.waitForCondition(
- () -> thread.taskManager().checkStateUpdater(
- mockTime.milliseconds(),
- topicPartitions -> mockConsumer.seekToBeginning(singleton(t1p1))
- ),
- 10 * 1000,
- "State updater never returned tasks.");
- }
+ TestUtils.waitForCondition(
+ () -> thread.taskManager().checkStateUpdater(
+ mockTime.milliseconds(),
+ topicPartitions -> mockConsumer.seekToBeginning(singleton(t1p1))
+ ),
+ 10 * 1000,
+ "State updater never returned tasks.");
+
Review Comment:
Trailing whitespace detected. Please remove the trailing whitespace on this line for consistency with the project's code style.
```suggestion
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java:
##########
@@ -1449,30 +1416,19 @@ public void shouldNotReturnDataAfterTaskMigrated(final boolean stateUpdaterEnabl
final TaskMigratedException taskMigratedException = new TaskMigratedException(
"Changelog restore found task migrated", new RuntimeException("restore task migrated"));
- ChangelogReader changelogReader = this.changelogReader;
- if (stateUpdaterEnabled) {
- when(taskManager.checkStateUpdater(anyLong(), any())).thenAnswer(answer -> {
- consumer.addRecord(new ConsumerRecord<>(topic1, 1, 11, new byte[0], new byte[0]));
- consumer.addRecord(new ConsumerRecord<>(topic1, 1, 12, new byte[1], new byte[0]));
+ final ChangelogReader changelogReader = this.changelogReader;
+ when(taskManager.checkStateUpdater(anyLong(), any())).thenAnswer(answer -> {
+ consumer.addRecord(new ConsumerRecord<>(topic1, 1, 11, new byte[0], new byte[0]));
+ consumer.addRecord(new ConsumerRecord<>(topic1, 1, 12, new byte[1], new byte[0]));
- throw taskMigratedException;
- });
- } else {
- changelogReader = new MockChangelogReader() {
- @Override
- public long restore(final Map<TaskId, Task> tasks) {
- consumer.addRecord(new ConsumerRecord<>(topic1, 1, 11, new byte[0], new byte[0]));
- consumer.addRecord(new ConsumerRecord<>(topic1, 1, 12, new byte[1], new byte[0]));
-
- throw taskMigratedException;
- }
- };
- }
+ throw taskMigratedException;
+ });
Review Comment:
Trailing whitespace detected. Please remove the trailing whitespace on this line for consistency with the project's code style.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java:
##########
@@ -1788,15 +1743,14 @@ public void shouldReinitializeRevivedTasksInAnyState(final boolean stateUpdaterE
// Now, we can handle the corruption
thread.taskManager().handleCorruption(taskCorruptedException.corruptedTasks());
- if (stateUpdaterEnabled) {
- TestUtils.waitForCondition(
- () -> thread.taskManager().checkStateUpdater(
- mockTime.milliseconds(),
- topicPartitions -> mockConsumer.seekToBeginning(singleton(t1p1))
- ),
- 10 * 1000,
- "State updater never returned tasks.");
- }
+ TestUtils.waitForCondition(
+ () -> thread.taskManager().checkStateUpdater(
+ mockTime.milliseconds(),
+ topicPartitions -> mockConsumer.seekToBeginning(singleton(t1p1))
+ ),
+ 10 * 1000,
+ "State updater never returned tasks.");
+
Review Comment:
Trailing whitespace detected. Please remove the trailing whitespace on this line for consistency with the project's code style.
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org