(batchprocessor): block requests until response is received #71
Conversation
moh-osman3
force-pushed
the
mohosman/batchprocessor
branch
from
2347913 to
bdbbabd
Compare
moh-osman3
changed the title
| } else { // waiter gets a complete response. | ||
| numItemsBefore += b.pending[0].numItems | ||
| if trigger == triggerTimeout { | ||
| err = multierr.Append(err, errTimedOut) |
There was a problem hiding this comment.
Hmm, this looks like we're making up an error, let's discuss. Is this necessary?
There was a problem hiding this comment.
Oh maybe I'm confused.. should the waiter be notified if their request times out? I thought the answer was yes, so that's why I'm letting the completeError indicate the timeout. Do you think timeout should simply be logged instead?
|
|
||
| numItemsBefore := b.totalSent | ||
| numItemsAfter := b.totalSent + sent | ||
| // The current batch can contain items from several different producers. Ensure each producer gets a response back. |
There was a problem hiding this comment.
Note: in a follow-on PR I'd like the export to happen in the background -- it's going to mean running through this loop before calling export, so we know the set of complete/pending responses that will be returned when the response comes back.
| var ( | ||
| // errTooManyBatchers is returned when the MetadataCardinalityLimit has been reached. | ||
| errTooManyBatchers = consumererror.NewPermanent(errors.New("too many batcher metadata-value combinations")) | ||
| errTimedOut = errors.New("processing items timed out") |
There was a problem hiding this comment.
Note: It looks to me like a non-error condition below, where you used this variable. Maybe we can remove it, let's discuss.
| respCh := make(chan error, 1) | ||
| // TODO: add a semaphore to only write to channel if sizeof(data) keeps | ||
| // us below some configured inflight byte limit. | ||
| sb.batcher.newItem <- dataItem{ |
There was a problem hiding this comment.
Note: not for now, but to consider: if the other side of this channel is being slow, this will block until after the context times out. This could instead be more defensive, like
select {
case sb.batcher.newItem <- dataItem{...}:
// ok
case <-ctx.Done():
return ctx.Err()
}
There was a problem hiding this comment.
Thanks for pointing this out! I went ahead and added the select statement you suggested
Part 2 for #80
This PR adds the batchprocessor to otel-arrow with a couple enhancements:
The main files to review are
batch_processor.goandbatch_processor_test.go