-
[OPIK-2654] [BE] Add retry logic and error handling for Redis stream subscriber (#3937)
frostbyte_neo 发布于
2025-11-06 09:33:46 +00:00 -
[OPIK-2654] [BE] Add retry logic and error handling for Redis stream subscriber
-
Revision 2: Fix getDeliveryCount compilation error
-
Revision 3: Use non-deprecated listPending API with StreamPendingRangeArgs
- Replaced deprecated listPending(String, StreamMessageId, StreamMessageId, int)
- Now using StreamPendingRangeArgs.groupName().startId().endId().count() syntax
- Updated unit test mock to use StreamPendingRangeArgs parameter
- All integration and unit tests passing
- Revision 4: Optimize delivery count queries to skip non-retryable errors
- Separate non-retryable failures first before querying delivery counts
- Only query Redis delivery counts for retryable failures
- Reduces unnecessary Redis calls for non-retryable errors
- All 28 tests passing (12 integration + 16 unit)
- Revision 5: Use Set for non-retryable exceptions and fix filter/flatMap logic
- Replace instanceof chain with Set for cleaner exception classification
- Fix inverted filter logic: return false for messages under retry limit
- Fix .map() to .flatMap() for ackAndRemoveMessages execution
- All 28 tests passing (12 integration + 16 unit)
- Revision 6: Simplify processEvent Javadoc and refine exception set
- Simplify exception handling Javadoc to reference NON_RETRYABLE_EXCEPTIONS set
- Change Set type from Throwable to RuntimeException for type safety
- Add ArrayIndexOutOfBoundsException to non-retryable set
- Remove IllegalMonitorStateException (rare edge case)
- Alphabetically sorted exception list maintained
- Revision 7: Add metrics for listPending operation
- Add listPendingErrors counter for error tracking
- Add listPendingTime histogram for timing measurement
- Initialize metrics in constructor following existing patterns
- Add error counter in onErrorResume and timing in doFinally
- All 28 tests passing (12 integration + 16 unit)
-
Revision 8: test improvements
-
Revision 9: Add lenient mocking for listPending in all applicable tests
-
Revision 10: Add import style guideline to code_style.mdc and remove duplication from testing.mdc
-
Revision 11: Add explicit verification for postProcessFailureMessages invocation
- Add shouldCallPostProcessFailureForNonRetryableErrors test
- Add shouldCallPostProcessFailureForRetryableErrors test
- Verify postProcessFailureMessages is called by checking:
- ack() called for non-retryable errors (immediately removed)
- listPending() called for retryable errors (delivery count query)
- Add Mockito verify() and atLeastOnce() imports
- All 28 tests passing (18 unit + 10 integration)
-
Revision 12: unit tests improvements
-
Revision 13: Add missing configuration properties to startup log
- Add claimIntervalRatio to log statement
- Add pendingMessageDuration to log statement
- Add maxRetries to log statement
- Now logs all StreamConfiguration properties except getCodec()
- All 28 tests passing (18 unit + 10 integration)
- Update apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriber.java
Co-authored-by: Copilot 175728472+Copilot@users.noreply.github.com
- Revision 14: Address PR review comments for code quality and clarity
- Extract shouldRemoveAfterMaxRetries() helper method for delivery count filtering
- Extract logAndExtractMaxRetriesFailure() helper method for max retries handling
- Update getDeliveryCount() javadoc to clarify PendingEntry::getLastTimeDelivered usage
Co-authored-by: Copilot 175728472+Copilot@users.noreply.github.com
下载附件
-