unknowntpo commented on code in PR #22309:
URL: https://github.com/apache/kafka/pull/22309#discussion_r3314898327
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java:
##########
@@ -1562,6 +1563,117 @@ public void
shouldSeekToBeginningForNonWindowedStoreWithoutCheckpoint() {
assertEquals(0L, consumer.position(tp), "Non-windowed store should
seek to beginning, not by timestamp");
}
+ @Test
+ public void shouldRejectInvalidRestoreBufferedRecordsPerPartition() {
+ final Properties properties = new Properties();
+
properties.put(StreamsConfig.RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG, 0);
+ assertThrows(ConfigException.class,
+ () -> new
StreamsConfig(StreamsTestUtils.getStreamsConfig("test-reader", properties)));
+ }
+
+ @Test
+ public void shouldPausePartitionWhenRestoreBufferCapIsReached() {
+ // standby with a committed-offset limit lets records pile up past the
limit
+ setupStandbyStateManager();
+ setupStoreMetadata();
+ setupStore();
+ // standby fixture has a null taskId, so the predicate looks up
tasks.get(null)
+ @SuppressWarnings("unchecked")
+ final Map<TaskId, Task> mockTasks = mock(Map.class);
+ when(mockTasks.get(null)).thenReturn(mock(Task.class));
+ when(mockTasks.containsKey(null)).thenReturn(true);
+ when(standbyStateManager.changelogAsSource(tp)).thenReturn(true);
+ when(storeMetadata.offset()).thenReturn(3L);
+ when(storeMetadata.endOffset()).thenReturn(20L);
+
+ final Properties properties = new Properties();
+
properties.put(StreamsConfig.RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG, 3);
+ properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
+ final StreamsConfig cappedConfig =
+ new StreamsConfig(StreamsTestUtils.getStreamsConfig("test-reader",
properties));
+ final StoreChangelogReader changelogReader =
+ new StoreChangelogReader(time, cappedConfig, logContext,
adminClient, consumer, callback, standbyListener);
+ changelogReader.transitToUpdateStandby();
+
+ consumer.updateBeginningOffsets(Collections.singletonMap(tp, 0L));
+ adminClient.updateConsumerGroupOffsets(Collections.singletonMap(tp,
7L));
+ changelogReader.register(tp, standbyStateManager);
+
+ changelogReader.restore(mockTasks);
+ assertEquals(Collections.emptySet(), consumer.paused());
+
+ // committed offset is 7, so offsets >=7 stay buffered
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 5L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 6L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 7L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 8L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 9L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 10L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 11L,
"key".getBytes(), "value".getBytes()));
+
+ // first tick fills the buffer, second tick sees it full and pauses
+ changelogReader.restore(mockTasks);
+ changelogReader.restore(mockTasks);
+
+
assertTrue(changelogReader.changelogMetadata(tp).bufferedRecords().size() >= 3);
Review Comment:
nit: Could we use a named local variable for
`RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG` here, and assert against it
instead of repeating the literal `3`?
```suggestion
final int restoreBufferedRecordsPerPartition = 3;
properties.put(StreamsConfig.RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG,
restoreBufferedRecordsPerPartition);
```
Then for the assertion line, use a second suggestion:
nit: Using the same named value here would make it clear this assertion is
tied to `RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG`, not an arbitrary
number.
```suggestion
assertTrue(changelogReader.changelogMetadata(tp).bufferedRecords().size() >=
restoreBufferedRecordsPerPartition);
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java:
##########
@@ -1562,6 +1563,117 @@ public void
shouldSeekToBeginningForNonWindowedStoreWithoutCheckpoint() {
assertEquals(0L, consumer.position(tp), "Non-windowed store should
seek to beginning, not by timestamp");
}
+ @Test
+ public void shouldRejectInvalidRestoreBufferedRecordsPerPartition() {
+ final Properties properties = new Properties();
+
properties.put(StreamsConfig.RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG, 0);
+ assertThrows(ConfigException.class,
+ () -> new
StreamsConfig(StreamsTestUtils.getStreamsConfig("test-reader", properties)));
+ }
+
+ @Test
+ public void shouldPausePartitionWhenRestoreBufferCapIsReached() {
+ // standby with a committed-offset limit lets records pile up past the
limit
+ setupStandbyStateManager();
+ setupStoreMetadata();
+ setupStore();
+ // standby fixture has a null taskId, so the predicate looks up
tasks.get(null)
+ @SuppressWarnings("unchecked")
+ final Map<TaskId, Task> mockTasks = mock(Map.class);
+ when(mockTasks.get(null)).thenReturn(mock(Task.class));
+ when(mockTasks.containsKey(null)).thenReturn(true);
+ when(standbyStateManager.changelogAsSource(tp)).thenReturn(true);
+ when(storeMetadata.offset()).thenReturn(3L);
+ when(storeMetadata.endOffset()).thenReturn(20L);
+
+ final Properties properties = new Properties();
+
properties.put(StreamsConfig.RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG, 3);
+ properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
+ final StreamsConfig cappedConfig =
+ new StreamsConfig(StreamsTestUtils.getStreamsConfig("test-reader",
properties));
+ final StoreChangelogReader changelogReader =
+ new StoreChangelogReader(time, cappedConfig, logContext,
adminClient, consumer, callback, standbyListener);
+ changelogReader.transitToUpdateStandby();
+
+ consumer.updateBeginningOffsets(Collections.singletonMap(tp, 0L));
+ adminClient.updateConsumerGroupOffsets(Collections.singletonMap(tp,
7L));
+ changelogReader.register(tp, standbyStateManager);
+
+ changelogReader.restore(mockTasks);
+ assertEquals(Collections.emptySet(), consumer.paused());
+
+ // committed offset is 7, so offsets >=7 stay buffered
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 5L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 6L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 7L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 8L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 9L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 10L,
"key".getBytes(), "value".getBytes()));
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 11L,
"key".getBytes(), "value".getBytes()));
+
+ // first tick fills the buffer, second tick sees it full and pauses
+ changelogReader.restore(mockTasks);
+ changelogReader.restore(mockTasks);
+
+
assertTrue(changelogReader.changelogMetadata(tp).bufferedRecords().size() >= 3);
+ assertEquals(Collections.singleton(tp), consumer.paused());
+ }
+
+ @Test
+ public void shouldResumePartitionOnceBufferDrainsBelowCap() {
+ setupStandbyStateManager();
+ setupStoreMetadata();
+ setupStore();
+ // standby fixture has a null taskId, so the predicate looks up
tasks.get(null)
+ @SuppressWarnings("unchecked")
+ final Map<TaskId, Task> mockTasks = mock(Map.class);
+ when(mockTasks.get(null)).thenReturn(mock(Task.class));
+ when(mockTasks.containsKey(null)).thenReturn(true);
+ when(standbyStateManager.changelogAsSource(tp)).thenReturn(true);
+ when(storeMetadata.offset()).thenReturn(3L);
+ when(storeMetadata.endOffset()).thenReturn(20L);
+
+ final long now = time.milliseconds();
+ final Properties properties = new Properties();
+
properties.put(StreamsConfig.RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG, 3);
+ properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
+ final StreamsConfig cappedConfig =
+ new StreamsConfig(StreamsTestUtils.getStreamsConfig("test-reader",
properties));
+ final StoreChangelogReader changelogReader =
+ new StoreChangelogReader(time, cappedConfig, logContext,
adminClient, consumer, callback, standbyListener);
+ changelogReader.transitToUpdateStandby();
+
+ consumer.updateBeginningOffsets(Collections.singletonMap(tp, 0L));
+ adminClient.updateConsumerGroupOffsets(Collections.singletonMap(tp,
7L));
+ changelogReader.register(tp, standbyStateManager);
+
+ changelogReader.restore(mockTasks);
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, 5L,
"key".getBytes(), "value".getBytes()));
Review Comment:
ditto, avoid hard coded partition index.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
