From ad7a41fa06cefa9564d54a3bc90b9aed8ae22dd7 Mon Sep 17 00:00:00 2001 From: HHans09 Date: Wed, 21 Sep 2022 13:57:21 -0400 Subject: [PATCH] FB-1646 Removed Debugf messages that does not make sense --- idk/kafka/source.go | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/idk/kafka/source.go b/idk/kafka/source.go index c19277c99..55fd4158e 100644 --- a/idk/kafka/source.go +++ b/idk/kafka/source.go @@ -210,7 +210,7 @@ func (r *Record) Commit(ctx context.Context) error { return errors.New("cannot commit a record that has already been committed") } section, remaining := r.src.spool[:idx-base], r.src.spool[idx-base:] - + sort.Slice(section, func(i, j int) bool { if *section[i].Topic != *section[j].Topic { return *section[i].Topic < *section[j].Topic @@ -231,7 +231,7 @@ func (r *Record) Commit(ctx context.Context) error { s = *x.Topic p = x.Partition } - + r.src.spool = remaining r.src.spoolBase = idx return nil @@ -299,13 +299,11 @@ func (s *Source) Open() error { return errors.Wrap(err, "new consumer") } - err = cl.SubscribeTopics(s.Topics, nil) if err != nil { return errors.Wrap(err, "subscribe topics") } - s.client = cl s.opened = true s.wg.Add(1)