Incorporate kafka tests into github work flow (#6)

1. Incorporate kafka tests into github workflow
2. Updated tests to assert that worker exits without error
3. Updated busy loop pause to respect context cancellation
4. Updated virtual partition write to respect context cancellation
5. Added debug logs for reader/processor exit
6. Removed dependency on github.com/errors package (no need for stack
trace info)
7. Added go report card
8. Corrected spelling mistakes
9. Updated coverage.sh script to echo each line
10. Updated compose.yml (renamed from docker-compose) to use
kafka+zookeeper instead of KRAFT implementation (in local usage)
11. Updated dependencies (confluent-kafka-go/otel)

.github/workflows/go.yml

@@ -32,7 +32,15 @@ jobs:
         run: |
           go mod download
 
+      - name: Run Kafka KRaft Broker
+        uses: spicyparrot/[email protected]
+        with:
+          kafka-version: "3.7.0"
+          kafka-topics: "example,1"
+
       - name: Test
+        env:
+          KAFKA_BOOTSTRAP_SERVER: ${{ env.kafka_runner_address }}:9092
         run: make cover
 
       - name: Upload coverage reports to Codecov

Makefile

@@ -1,20 +1,17 @@
 # Directories containing independent Go modules.
 MODULE_DIRS = .
 
-.PHONY: test-no-setup
-test-no-setup:
-	./coverage.sh
 
 .PHONY: setup-test
 setup-test:
-	docker compose -p $$RANDOM -f ./example/docker-compose.yaml up -d
+	docker compose -p $$RANDOM -f ./example/compose.yaml up -d
 
 .PHONY: test-local
-test-local: setup-test test-no-setup
+test-local: setup-test cover
 
 .PHONY: cover
 cover:
-	./coverage.sh
+	export GO_TAGS=--tags=integration; ./coverage.sh --tags=integration
 
 .PHONY: example-producer
 example-producer:

README.md

@@ -3,7 +3,7 @@
 [![License](https://img.shields.io/github/license/zillow/zkafka)](https://github.com/zillow/zkafka/blob/main/LICENSE)
 [![GitHub Actions](https://github.com/zillow/zkafka/actions/workflows/go.yml/badge.svg)](https://github.com/zillow/zkafka/actions/workflows/go.yml)
 [![Codecov](https://codecov.io/gh/zillow/zkafka/branch/main/graph/badge.svg?token=STRT8T67YP)](https://codecov.io/gh/zillow/zkafka)
-
+[![Go Report Card](https://goreportcard.com/badge/github.com/zillow/zkafka)](https://goreportcard.com/report/github.com/zillow/zkafka)
 
 ## Install
 
@@ -93,7 +93,7 @@ client because that value is explicitly set to true after reading of the Additio
   "KafkaTopicConfig": {
     "Topic": "KafkaTopicName",
     "BootstrapServers": [
-      "localhost:9093"
+      "localhost:9092"
     ],
     // translates to librdkafka value "bootstrap.servers"
     // specify ad hoc configuration values which don't have a strongly typed version in the TopicConfig struct.

commitmgr_test.go

@@ -197,8 +197,8 @@ func Test_commitMgr_PerPartitionDataStructuresBuiltUpConcurrentlyCorrectly(t *te
 	require.Equal(t, int64(0), mgr.inWorkCount, "expected inWorkCount to be empty")
 }
 
-// Test_commitMgr_mutex_ShouldReturnReferenceToSameMutexForSamePartition tests for a race condition (based on bugfound)
-// where two goroutines calling this method received distinct mutexes (which isn't correct for syncronization purposes).
+// Test_commitMgr_mutex_ShouldReturnReferenceToSameMutexForSamePartition tests for a race condition (based on bug found)
+// where two goroutines calling this method received distinct mutexes (which isn't correct for synchronization purposes).
 // Try a large amount of times to access the mutex for a particular partition. Always should return same pointer
 func Test_commitMgr_mutex_ShouldReturnReferenceToSameMutexForSamePartition(t *testing.T) {
 	defer recoverThenFail(t)

coverage.sh

@@ -1,4 +1,8 @@
 #!/usr/bin/env bash
+set -x
+
+go_tags=$GO_TAGS
+go_tags="${go_tags:---tags=unit}"
 
 # golang packages that will be used for either testing or will be assessed for coverage
 pck1=github.com/zillow/zkafka
@@ -22,11 +26,11 @@ function quit() {
 }
 # change to example directory for execution (because it uses hardcoded filepaths, and the testable
 # examples don't work when executed outside of that directory
-go test -c -coverpkg=$pck1 -covermode=atomic -o "$root_res" $pck1
+go test $go_tags -c -coverpkg=$pck1 -covermode=atomic -o "$root_res" $pck1
 # convert binary to go formatted
 go tool test2json -t "$root_res" -test.v -test.coverprofile "$root_out"
 
-go test -c -coverpkg=$pck1 -covermode=atomic -o "$source_res" $pck2
+go test $go_tags -c -coverpkg=$pck1 -covermode=atomic -o "$source_res" $pck2
 go tool test2json -t "$source_res" -test.v -test.coverprofile "$source_out"
 
 # delete aggregate file

example/compose.yaml

@@ -0,0 +1,24 @@
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:latest
+    container_name: zkafka-zookeeper
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    ports:
+      - "22181:2181"
+  kafka:
+    image: confluentinc/cp-kafka:latest
+    container_name: zkafka-broker
+    depends_on:
+      - zookeeper
+    ports:
+      - "29092:29092"
+      - "9092:9092"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

example/consumer/consumer.go

@@ -12,7 +12,7 @@ import (
 func main() {
 	// configure broker connectivity options with zkafka.Config
 	cfg := zkafka.Config{
-		BootstrapServers: []string{"localhost:9093"},
+		BootstrapServers: []string{"localhost:9092"},
 	}
 
 	// configure consumer options  with zkafka.ConsumerTopicConfig. See zkafka for full option values

example/producer/producer.go

@@ -14,7 +14,7 @@ import (
 func main() {
 	ctx := context.Background()
 	writer, err := zkafka.NewClient(zkafka.Config{
-		BootstrapServers: []string{"localhost:9093"},
+		BootstrapServers: []string{"localhost:9092"},
 	}).Writer(ctx, zkafka.ProducerTopicConfig{
 		ClientID:  "example",
 		Topic:     "two-multi-partition",

example/worker/worker.go

@@ -17,7 +17,7 @@ import (
 func main() {
 	ctx := context.Background()
 	client := zkafka.NewClient(zkafka.Config{
-		BootstrapServers: []string{"localhost:9093"},
+		BootstrapServers: []string{"localhost:9092"},
 	},
 		zkafka.LoggerOption(stdLogger{}),
 	)

go.mod

@@ -3,16 +3,15 @@ module github.com/zillow/zkafka
 go 1.22
 
 require (
-	github.com/confluentinc/confluent-kafka-go/v2 v2.4.0
+	github.com/confluentinc/confluent-kafka-go/v2 v2.5.0
 	github.com/golang/mock v1.6.0
 	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.6.0
-	github.com/pkg/errors v0.9.1
 	github.com/sony/gobreaker v1.0.0
 	github.com/stretchr/testify v1.9.0
 	github.com/zillow/zfmt v1.0.1
-	go.opentelemetry.io/otel v1.27.0
-	go.opentelemetry.io/otel/trace v1.27.0
+	go.opentelemetry.io/otel v1.28.0
+	go.opentelemetry.io/otel/trace v1.28.0
 	golang.org/x/sync v0.7.0
 )
 
@@ -24,7 +23,7 @@ require (
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/heetch/avro v0.4.5 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	go.opentelemetry.io/otel/metric v1.27.0 // indirect
+	go.opentelemetry.io/otel/metric v1.28.0 // indirect
 	google.golang.org/protobuf v1.34.2 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )