fixes

.github/workflows/test.yaml

@@ -38,7 +38,7 @@ jobs:
         run: go vet ./...
 
       - name: Test
-        run: go test -bench=. -covermode=count -coverprofile=profile_full.cov -coverpkg=github.com/msales/streams/... ./...
+        run: go test -race -bench=. -covermode=count -coverprofile=profile_full.cov -coverpkg=github.com/msales/streams/... ./...
 
       - name: Goveralls
         run: |

kafka/source.go

@@ -302,7 +302,7 @@ func (s *Source) Commit(v interface{}) error {
 	// If commit strategy is not CommitAuto, session should perform global, synchronous commit of current marked offsets.
 	if s.commitStrategy != CommitAuto {
 		s.session.Commit()
-		runtime.Gosched() // If any error from consumer side happens after Commiting, it will be read and s.lastErr will be set.
+		runtime.Gosched() // If any error from consumer side happens after Committing, it will be read and sent to errors channel.
 	}
 
 	select {
@@ -388,7 +388,7 @@ func (s *Source) runConsumerGroup(ctx context.Context, topic string) {
 
 	for {
 		err := s.consumer.Consume(ctx, []string{topic}, s)
-		if ctx.Err() == context.Canceled { // This is the proper way to end the consumption.
+		if errors.Is(ctx.Err(), context.Canceled) { // This is the proper way to end the consumption.
 			return
 		}
 		if err == nil {

kafka/source_test.go

@@ -545,8 +545,6 @@ func TestSource_Commit_Manual_ReturnError(t *testing.T) {
 	_, err = s.Consume()
 	require.NoError(t, err)
 
-	time.Sleep(100 * time.Millisecond)
-
 	err = s.Commit(meta)
 
 	assert.Error(t, err)