You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "muralibasani (via GitHub)" <gi...@apache.org> on 2026/05/18 16:07:47 UTC
[PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
muralibasani opened a new pull request, #22309:
URL: https://github.com/apache/kafka/pull/22309
Ref : https://issues.apache.org/jira/browse/KAFKA-20589
Add restore.buffered.records.per.partition config and enforce cap in StoreChangelogReader
--
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
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #22309:
URL: https://github.com/apache/kafka/pull/22309#issuecomment-4566290238
@unknowntpo thanks for the review.
Pushed changes based on the comments.
--
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
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
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 hard-coded `3`? Its more readable.
```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);
```
--
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
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #22309:
URL: https://github.com/apache/kafka/pull/22309#discussion_r3314781149
##########
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()));
Review Comment:
Same here, can we use `tp.partition()` instead of hard-coding `0`?
```suggestion
consumer.addRecord(new ConsumerRecord<>(topicName, tp.partition(), 6L, "key".getBytes(), "value".getBytes()));
```
--
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
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
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 hard-coded `3`? Its more readable.
```suggestion
assertTrue(changelogReader.changelogMetadata(tp).bufferedRecords().size() >= restoreBufferedRecordsPerPartition);
```
--
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
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #22309:
URL: https://github.com/apache/kafka/pull/22309#discussion_r3314791637
##########
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()));
Review Comment:
nit: Can we use `tp.partition()` here to keep the record partition tied to the `TopicPartition` used by the test?
```suggestion
consumer.addRecord(new ConsumerRecord<>(topicName, tp.partition(), 5L, "key".getBytes(), "value".getBytes()));
```
--
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
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #22309:
URL: https://github.com/apache/kafka/pull/22309#discussion_r3314781149
##########
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()));
Review Comment:
Same here, can we use `tp.partition()` instead of hard-coding `0`?
```suggestion
consumer.addRecord(new ConsumerRecord<>(topicName, tp.partition(), 6L, "key".getBytes(), "value".getBytes()));
```
--
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
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22309:
URL: https://github.com/apache/kafka/pull/22309#issuecomment-4540047556
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on code in PR #22309:
URL: https://github.com/apache/kafka/pull/22309#discussion_r3314911882
##########
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:
same as https://github.com/apache/kafka/pull/22309#discussion_r3314791637, 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
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 hard-coded `3`? Its more readable.
```suggestion
final int restoreBufferedRecordsPerPartition = 3;
properties.put(StreamsConfig.RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG, restoreBufferedRecordsPerPartition);
```
Then for the assertion line:
```suggestion
assertTrue(changelogReader.changelogMetadata(tp).bufferedRecords().size() >= restoreBufferedRecordsPerPartition);
```
--
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
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
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 introduce a named local variable for `RESTORE_BUFFERED_RECORDS_PER_PARTITION_CONFIG`, .e.g `restoreBufferedRecordsPerPartition`, and assert against it instead of hard-coded `3`? Its more readable.
```suggestion
assertTrue(changelogReader.changelogMetadata(tp).bufferedRecords().size() >= restoreBufferedRecordsPerPartition);
```
--
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
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "unknowntpo (via GitHub)" <gi...@apache.org>.
unknowntpo commented on PR #22309:
URL: https://github.com/apache/kafka/pull/22309#issuecomment-4569745381
LGTM.
--
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
Re: [PR] KAFKA-20589 : Adding new streams config for restore path to handle memory issues [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22309:
URL: https://github.com/apache/kafka/pull/22309#issuecomment-6307504119
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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