fix(interactive): Fix several sparodic bugs in groot #4595
Conversation
There was a problem hiding this comment.
Pull Request Overview
This PR fixes several sporadic bugs in the groot project by updating the Kafka log reading API usage and modifying RocksDB operations.
- Update test and processing code to use ConsumerRecord based API instead of deprecated ReadLogEntry.
- Adjust KafkaLogReader error handling in polling and modify RocksDB background work cancellation with a blocking sleep.
- Update Cargo.toml to include an additional RocksDB feature.
Reviewed Changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| interactive_engine/groot-server/src/test/java/com/alibaba/graphscope/groot/tests/common/wal/kafka/KafkaWalTest.java | Updated tests to use ConsumerRecord API instead of ReadLogEntry. |
| interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/wal/kafka/KafkaLogReader.java | Modified reading logic with try-catch handling for Kafka polling and updated offset check. |
| interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/store/KafkaProcessor.java | Adapted log processing to use ConsumerRecord. |
| interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/frontend/write/KafkaAppender.java | Updated log reading code to use the new API. |
| interactive_engine/executor/store/groot/src/db/storage/rocksdb.rs | Revised RocksDB handling by canceling background work and introducing a sleep delay. |
| interactive_engine/executor/store/groot/src/db/graph/store.rs | Tweaked consistency check in snapshot guard with a tolerance range. |
| interactive_engine/executor/store/groot/Cargo.toml | Enabled additional RocksDB feature to support multi-threaded column families. |
Comments suppressed due to low confidence (1)
interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/wal/kafka/KafkaLogReader.java:134
- The catch block in the readNextRecord method adjusts the consumer offset by moving to cur + 1 on a poll failure, which may inadvertently skip records. Please verify that this behavior is intentional and that it does not compromise message ordering or data integrity.
} catch (KafkaException ex) {
| std::thread::sleep(Duration::from_secs(30)); | ||
| drop(prev); |
There was a problem hiding this comment.
The introduction of a blocking sleep for 30 seconds after canceling background work can significantly impact performance. Consider whether this delay is necessary or if a non-blocking or asynchronous approach might be more appropriate.
| std::thread::sleep(Duration::from_secs(30)); | |
| drop(prev); | |
| let prev_clone = prev.clone(); | |
| std::thread::spawn(move || { | |
| std::thread::sleep(Duration::from_secs(30)); | |
| drop(prev_clone); | |
| }); |
|
Please check the preview of the documentation changes at |
|
/cc @yecol @sighingnow, this issus/pr has had no activity for a long time, please help to review the status and assign people to work on it. |
Signed-off-by: siyuan0322 <[email protected]>
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #4595 +/- ##
==========================================
- Coverage 35.01% 33.02% -2.00%
==========================================
Files 127 127
Lines 13299 13299
==========================================
- Hits 4657 4392 -265
- Misses 8642 8907 +265 see 12 files with indirect coverage changes Continue to review full report in Codecov by Sentry.
🚀 New features to boost your workflow:
|
2 participants
No description provided.