Differentiate producer/consumer configuration (#16)

The producer/consumer code was sharing the same configuration, which
means the producer was being initialized as a consumer as well. If the
producer starts first, it joins the same group and is assigned the only
partition for the topic. When the consumer eventually comes online, it
joins the group but is not assigned the partition, appearing to be
stuck. This patch bifurcates the configuration on whether or not the
Kafka client being stood up should be a consumer. Also, the PR adds a
few hygienic options that should always be set for consumers (read
committed, require stable offsets).

cmd/bufstream-demo-consume/main.go

@@ -26,7 +26,7 @@ func main() {
 }
 
 func run(ctx context.Context, config app.Config) error {
-	client, err := kafka.NewKafkaClient(config.Kafka)
+	client, err := kafka.NewKafkaClient(config.Kafka, true)
 	if err != nil {
 		return err
 	}

cmd/bufstream-demo-produce/main.go

@@ -33,7 +33,7 @@ func main() {
 }
 
 func run(ctx context.Context, config app.Config) error {
-	client, err := kafka.NewKafkaClient(config.Kafka)
+	client, err := kafka.NewKafkaClient(config.Kafka, false)
 	if err != nil {
 		return err
 	}

pkg/kafka/kafka.go

@@ -28,14 +28,21 @@ type Config struct {
 }
 
 // NewKafkaClient returns a new franz-go Kafka Client for the given Config.
-func NewKafkaClient(config Config) (*kgo.Client, error) {
+func NewKafkaClient(config Config, consumer bool) (*kgo.Client, error) {
 	opts := []kgo.Opt{
 		kgo.SeedBrokers(config.BootstrapServers...),
-		kgo.ConsumerGroup(config.Group),
-		kgo.ConsumeTopics(config.Topic),
 		kgo.ClientID(config.ClientID),
 		kgo.AllowAutoTopicCreation(),
-		kgo.FetchMaxWait(time.Second),
+	}
+
+	if consumer {
+		opts = append(opts,
+			kgo.ConsumerGroup(config.Group),
+			kgo.ConsumeTopics(config.Topic),
+			kgo.FetchMaxWait(time.Second),
+			kgo.FetchIsolationLevel(kgo.ReadCommitted()),
+			kgo.RequireStableFetchOffsets(),
+		)
 	}
 
 	if config.RootCAPath != "" {