enhance: add broadcast for streaming service #39020
Conversation
mergify
bot
added
dco-passed
|
@chyezh go-sdk check failed, comment |
chyezh
force-pushed
the
enhance_add_broadcast_for_msg_system
branch
2 times, most recently
from
adc133d to
1292957
Compare
|
@chyezh E2e jenkins job failed, comment |
chyezh
force-pushed
the
enhance_add_broadcast_for_msg_system
branch
2 times, most recently
from
129b3bd to
9e1bfb5
Compare
|
@chyezh go-sdk check failed, comment |
|
@chyezh cpp-unit-test check failed, comment |
|
rerun cpp-unit-test |
|
rerun go-sdk |
- Add new rpc for transfer broadcast to streaming coord - Add broadcast service at streaming coord to make broadcast message sent Signed-off-by: chyezh <[email protected]>
chyezh
force-pushed
the
enhance_add_broadcast_for_msg_system
branch
from
9e1bfb5 to
4fe6647
Compare
|
|
||
| func (c *catalog) SaveBroadcastTask(ctx context.Context, task *streamingpb.BroadcastTask) error { | ||
| key := buildBroadcastTaskPath(task.TaskId) | ||
| if task.State == streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_DONE { |
There was a problem hiding this comment.
Use the RemoveBroadcastTask method to delete metadata from the catalog?
There was a problem hiding this comment.
I want to keep only state transfer semantic at catalog layer.
|
|
||
| // Start n workers to handle the broadcast task. | ||
| wg := sync.WaitGroup{} | ||
| for i := 0; i < 4; i++ { |
There was a problem hiding this comment.
Use the CPU count instead of hardcoding the number 4.
| case task := <-b.backoffChan: | ||
| // task is backoff, push it into backoff queue to make a delay retry. | ||
| b.backoffs.Push(task) | ||
| case <-nextBackOff: |
There was a problem hiding this comment.
Reinsert the backoff task at the front of the pending queue; otherwise, consuming messages waiting for Barrie will take a long time.
|
|
||
| // Poll polls the task, return nil if the task is done, otherwise not done. | ||
| // Poll can be repeated called until the task is done. | ||
| func (b *broadcastTask) Poll(ctx context.Context, operator AppendOperator) error { |
There was a problem hiding this comment.
The method is somewhat confusingly named "poll," as it actually executes the task. so suggest separating the task's state check and its execution into two distinct methods.
| if len(newPendings) == 0 { | ||
| b.future.Set(&types.BroadcastAppendResult{AppendResults: b.appendResult}) | ||
| } | ||
| b.logger.Info("broadcast task make a new broadcast done", zap.Int("pendingMessages", len(b.pendingMessages))) |
There was a problem hiding this comment.
use "backoff messages " instead of "pendingMessages"
|
Address the issue in the next PR. |
|
[APPROVALNOTIFIER] This PR is APPROVED Approval requirements bypassed by manually added approval. This pull-request has been approved by: chyezh The full list of commands accepted by this bot can be found here. The pull request process is described here
Needs approval from an approver in each of these files:
Approvers can indicate their approval by writing |
issue: #38399 pr: #39020 - Add new rpc for transfer broadcast to streaming coord - Add broadcast service at streaming coord to make broadcast message sent automicly also cherry pick the pr #38400 --------- Signed-off-by: chyezh <[email protected]>
issue: #38399