You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mjsax (via GitHub)" <gi...@apache.org> on 2025/01/11 03:05:41 UTC
[PR] KAFKA-18479: RocksDBTimeOrderedKeyValueBuffer not initialized correctly [kafka]
mjsax opened a new pull request, #18490:
URL: https://github.com/apache/kafka/pull/18490
RocksDBTimeOrderedKeyValueBuffer is not initialize with serdes provides via Joined, but always uses serdes from StreamsConfig.
--
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-18479: RocksDBTimeOrderedKeyValueBuffer not initialized correctly [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #18490:
URL: https://github.com/apache/kafka/pull/18490#issuecomment-2593802638
Merged to `trunk` and cherry-picked to `4.0` and `3.9` branches. `3.9` was already tricky (some conflicts) but `3.8` was even more tricky to cherry-pick, so I did not push directly but opened a PR: https://github.com/apache/kafka/pull/18564
--
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-18479: RocksDBTimeOrderedKeyValueBuffer not initialized correctly [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #18490:
URL: https://github.com/apache/kafka/pull/18490#discussion_r1911865335
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/StreamTableJoinWithGraceIntegrationTest.java:
##########
@@ -52,13 +54,13 @@ public class StreamTableJoinWithGraceIntegrationTest extends AbstractJoinIntegra
@ValueSource(booleans = {true, false})
public void testInnerWithVersionedStore(final boolean cacheEnabled) {
final StreamsBuilder builder = new StreamsBuilder();
- final KStream<Long, String> leftStream = builder.stream(INPUT_TOPIC_LEFT);
- final KTable<Long, String> rightTable = builder.table(INPUT_TOPIC_RIGHT, Materialized.as(
+ final KStream<Long, String> leftStream = builder.stream(INPUT_TOPIC_LEFT, Consumed.with(Serdes.Long(), Serdes.String()));
+ final KTable<Long, String> rightTable = builder.table(INPUT_TOPIC_RIGHT, Consumed.with(Serdes.Long(), Serdes.String()), Materialized.as(
Stores.persistentVersionedKeyValueStore(STORE_NAME, Duration.ofMinutes(5))));
- final Properties streamsConfig = setupConfigsAndUtils(cacheEnabled);
+ final Properties streamsConfig = setupConfigsAndUtils(cacheEnabled, false);
Review Comment:
To test `Joined` we use `false` here
--
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-18479: RocksDBTimeOrderedKeyValueBuffer not initialized correctly [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #18490:
URL: https://github.com/apache/kafka/pull/18490#discussion_r1911865862
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java:
##########
@@ -62,18 +62,16 @@ public class RocksDBTimeOrderedKeyValueBufferTest {
@SuppressWarnings({"rawtypes", "unchecked"})
@BeforeEach
public void setUp() {
- when(serdeGetter.keySerde()).thenReturn((Serde) new Serdes.StringSerde());
- when(serdeGetter.valueSerde()).thenReturn((Serde) new Serdes.StringSerde());
final Metrics metrics = new Metrics();
offset = 0;
streamsMetrics = new StreamsMetricsImpl(metrics, "test-client", "processId", new MockTime());
context = new MockInternalProcessorContext<>(StreamsTestUtils.getStreamsConfig(), new TaskId(0, 0), TestUtils.tempDirectory());
}
- private void createBuffer(final Duration grace) {
+ private void createBuffer(final Duration grace, final Serde<String> serde) {
Review Comment:
"randomizing" this test, to cover both code path.
--
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-18479: RocksDBTimeOrderedKeyValueBuffer not initialized correctly [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #18490:
URL: https://github.com/apache/kafka/pull/18490#discussion_r1911865599
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/StreamTableJoinWithGraceIntegrationTest.java:
##########
@@ -96,7 +98,7 @@ public void testLeftWithVersionedStore(final boolean cacheEnabled) {
final KStream<Long, String> leftStream = builder.stream(INPUT_TOPIC_LEFT);
final KTable<Long, String> rightTable = builder.table(INPUT_TOPIC_RIGHT, Materialized.as(
Stores.persistentVersionedKeyValueStore(STORE_NAME, Duration.ofMinutes(5))));
- final Properties streamsConfig = setupConfigsAndUtils(cacheEnabled);
+ final Properties streamsConfig = setupConfigsAndUtils(cacheEnabled, true);
Review Comment:
For `leftJoin` we stick with using `StreamsConfig` serde to test both setups.
--
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-18479: RocksDBTimeOrderedKeyValueBuffer not initialized correctly [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #18490:
URL: https://github.com/apache/kafka/pull/18490
--
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-18479: RocksDBTimeOrderedKeyValueBuffer not initialized correctly [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #18490:
URL: https://github.com/apache/kafka/pull/18490#discussion_r1911865297
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java:
##########
@@ -260,16 +266,13 @@ void runSelfJoinTestWithDriver(
private void checkQueryableStore(final String queryableName, final TestRecord<Long, String> expectedFinalResult, final TopologyTestDriver driver) {
final ReadOnlyKeyValueStore<Long, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore(queryableName);
- final KeyValueIterator<Long, ValueAndTimestamp<String>> all = store.all();
- final KeyValue<Long, ValueAndTimestamp<String>> onlyEntry = all.next();
+ try (final KeyValueIterator<Long, ValueAndTimestamp<String>> all = store.all()) {
Review Comment:
Side cleanup
--
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-18479: RocksDBTimeOrderedKeyValueBuffer not initialized correctly [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18490:
URL: https://github.com/apache/kafka/pull/18490#discussion_r1918297205
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java:
##########
@@ -62,18 +62,16 @@ public class RocksDBTimeOrderedKeyValueBufferTest {
@SuppressWarnings({"rawtypes", "unchecked"})
Review Comment:
Sorry for the delayed feedback. To address the following warnings, could you please move these suppression annotations to the class-level?
```
> Task :streams:compileTestJava
Note: Some input files use or override a deprecated API.
Note: Recompile with -Xlint:deprecation for details.
Note: /home/chia7712/project/kafka/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java uses unchecked or unsafe operations.
Note: Recompile with -Xlint:unchecked for details.
```
--
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-18479: RocksDBTimeOrderedKeyValueBuffer not initialized correctly [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #18490:
URL: https://github.com/apache/kafka/pull/18490#discussion_r1918974529
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java:
##########
@@ -62,18 +62,16 @@ public class RocksDBTimeOrderedKeyValueBufferTest {
@SuppressWarnings({"rawtypes", "unchecked"})
Review Comment:
Nice catch! -- I think we should _never_ use class level suppression though. Opened a PR to suppress for the individual methods: https://github.com/apache/kafka/pull/18580
--
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