Skip to content

Commit

Permalink
kgo: patch HookFetchRecordUnbuffered
Browse files Browse the repository at this point in the history
Previously, if a hook implemented both HookFetchRecordBuffered and
HookFetchRecordUnbuffered, then HookFetchRecordUnbuffered would never
be called. The type switch in this logic would always match the buffered
case and never call unbuffered.

Reorganizing the conditionals a little bit ensures that the correct hook
is always called.
  • Loading branch information
twmb committed Jan 22, 2023
1 parent 3868806 commit 2a37df9
Showing 1 changed file with 6 additions and 6 deletions.
12 changes: 6 additions & 6 deletions pkg/kgo/source.go
Original file line number Diff line number Diff line change
Expand Up @@ -289,9 +289,9 @@ type bufferedFetch struct {

func (s *source) hook(f *Fetch, buffered, polled bool) {
s.cl.cfg.hooks.each(func(h Hook) {
switch h := h.(type) {
case HookFetchRecordBuffered:
if !buffered {
if buffered {
h, ok := h.(HookFetchRecordBuffered)
if !ok {
return
}
for i := range f.Topics {
Expand All @@ -303,9 +303,9 @@ func (s *source) hook(f *Fetch, buffered, polled bool) {
}
}
}

case HookFetchRecordUnbuffered:
if buffered {
} else {
h, ok := h.(HookFetchRecordUnbuffered)
if !ok {
return
}
for i := range f.Topics {
Expand Down

0 comments on commit 2a37df9

Please sign in to comment.