enhance: timetick interceptor implementation #34238
Conversation
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: chyezh The full list of commands accepted by this bot can be found here.
Needs approval from an approver in each of these files:
Approvers can indicate their approval by writing |
mergify
bot
added
dco-passed
chyezh
force-pushed
the
feat_streaming_service_timetick_interceptor
branch
from
1ce0cfb
to
429774f
Compare
| const ( | ||
| ServiceMethodPrefix = "/milvus.proto.log" | ||
| InitialTerm = int64(-1) | ||
| ) | ||
|
|
||
| func NewDeliverAll() *DeliverPolicy { |
chyezh
force-pushed
the
feat_streaming_service_timetick_interceptor
branch
from
429774f
to
fdeada5
Compare
chyezh
force-pushed
the
feat_streaming_service_timetick_interceptor
branch
from
fdeada5
to
46bbe8c
Compare
|
@chyezh E2e jenkins job failed, comment |
Signed-off-by: chyezh <chyezh@outlook.com>
chyezh
force-pushed
the
feat_streaming_service_timetick_interceptor
branch
from
46bbe8c
to
915c301
Compare
|
@chyezh E2e jenkins job failed, comment |
|
/run-cpu-e2e |
|
@chyezh E2e jenkins job failed, comment |
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #34238 +/- ##
==========================================
- Coverage 84.61% 80.87% -3.74%
==========================================
Files 817 1096 +279
Lines 113473 137783 +24310
==========================================
+ Hits 96013 111438 +15425
- Misses 13238 22116 +8878
- Partials 4222 4229 +7
|
| @@ -0,0 +1,76 @@ | |||
| package resource | |||
There was a problem hiding this comment.
resource package is used to access the streamingnode global singleton.
| for _, opt := range opts { | ||
| opt(ta.detail) | ||
| } | ||
| ta.acknowledged.Store(true) |
There was a problem hiding this comment.
acknowledged should not be true if the error is not nil with ack option
There was a problem hiding this comment.
The acknowledgement here is for timetick manager.
All allocated timetick must be acked, Otherwise, the timetick interceptor based on heap got stuck.
The error here need to be handled by outside AckDetail management, but current behaviour is just ignored, and drop the illegal message at scanner side.
| } | ||
|
|
||
| // Construct time tick message. | ||
| msg, err := newTimeTickMsg(impl.ackDetails.LastAllAcknowledgedTimestamp(), impl.sourceID) |
There was a problem hiding this comment.
Here, we could ignore the error acknowledgement, may be skip some timtick sent, but I think it's make no sense.
Just drop the illegal message at scanner side based on timetick rule is best error-handling.
| logger.Info("start to sync time tick...") | ||
| defer logger.Info("sync time tick stopped") | ||
|
|
||
| // Send first timetick message to wal before interceptor is ready. |
There was a problem hiding this comment.
Move the following code snippet into a separate method named tryToSyncTimeTickUntilReady.
|
/lgtm |
issue: milvus-io#33285 - optimize the message package - add interceptor package to achieve append operation intercepting. - add timetick interceptor to attach timetick properties for message. - add timetick background task to send timetick message. Signed-off-by: chyezh <chyezh@outlook.com>
issue: #33285