Failing to run examples

[CG-SS]

When trying to run the very first example leads to the following error:

Error creating topic manager: Error creating the kafka client: kafka: invalid configuration (Consumer.Group.Rebalance.GroupStrategies and Consumer.Group.Rebalance.Strategy cannot be set at the same time).

Seems like the defaultConfig might be out of date?

[norbertklawikowski]

Hi @CG-SS,
I wasn't able to reproduce the error (from the current master, using make start from the examples directory before running the first example).
Which Kafka Version did you use? Sounds like there could be some incompatibility, like you already assumed.

[CG-SS]

This was done using the version 3.3.1 of Apache Kafka, however, I didn't try running the example via the make start command. What I did was copied the first example's code and ran it while running Apache Kafka locally.

[Saeed903]

I want run 3-messages example , but when run , first error is collector-table topic not exist, then I create topic in kafka cluster and run again new error is (when run: go run cmd/processor/main.go -collector ) :
error consuming from group consumer: 2 errors occurred:
* error setting up (partition=1): Setup failed. Cannot start processor for partition 0: 1 error occurred:
* kafka tells us there's no message in the topic, but our cache has one. The table might be gone. Try to delete your local cache! Topic collector-table, partition 0, hwm 0, local offset 369

[Derek-Hardy]

@Saeed903 may I ask the version of Kafka you are using that produce this error?

[CG-SS]

I got it to work by just copying and pasting the default config but without the Consumer.Group.Rebalance.Strategy defined, as the following:

        config := sarama.NewConfig()
	// consumer configuration
	config.Consumer.Return.Errors = true
	config.Consumer.MaxProcessingTime = defaultMaxProcessingTime
	// this configures the initial offset for streams. Tables are always
	// consumed from OffsetOldest.
	config.Consumer.Offsets.Initial = sarama.OffsetNewest
	// producer configuration
	config.Producer.RequiredAcks = sarama.WaitForLocal
	config.Producer.Compression = sarama.CompressionSnappy
	config.Producer.Flush.Frequency = defaultFlushFrequency
	config.Producer.Flush.Bytes = defaultFlushBytes
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Producer.Retry.Max = defaultProducerMaxRetries
	// since the emitter only emits one message, we need to tell the processor
	// to read from the beginning
	// As the processor is slower to start than the emitter, it would not consume the first
	// message otherwise.
	// In production systems however, check whether you really want to read the whole topic on first start, which
	// can be a lot of messages.
	config.Consumer.Offsets.Initial = sarama.OffsetOldest
	goka.ReplaceGlobalConfig(config)

Where the consts are:

const (
	// time sarama-cluster assumes the processing of an event may take
	defaultMaxProcessingTime = 1 * time.Second

	// producer flush configuration
	defaultFlushFrequency     = 100 * time.Millisecond
	defaultFlushBytes         = 64 * 1024
	defaultProducerMaxRetries = 10
)

Maybe I missed it, but I don't see anywhere in the docs stating which version of Kafka goka aims to support by default.

[frairon]

Seems like it was a sarama issue and was fixed here: IBM/sarama#2351

This PR also updates dependencies, and trying to run the examples there does work without problems, so I'm assuming it's fixed from sarama side and we can close this.