Tester deadlock workaround

tester/queue.go

@@ -82,7 +82,8 @@ func (q *queue) messagesFromOffset(offset int64) []*message {
 func (q *queue) waitConsumersInit() {
 	logger.Printf("Consumers in Queue %s", q.topic)
 	for cons := range q.groupConsumers {
-		logger.Printf("waiting for group consumer %s to be running", cons.queue.topic)
+		logger.Printf("waiting for group consumer %s to be running or killed (state=%v)", cons.queue.topic, cons.state.State())
+
 		select {
 		case <-cons.state.WaitForState(killed):
 			log.Printf("At least one consumer was killed. No point in waiting for it")

tester/queueconsumer.go

@@ -2,6 +2,7 @@ package tester
 
 import (
 	"fmt"
+	"log"
 	"sync"
 	"time"
 
@@ -14,6 +15,7 @@ type consumer struct {
 	subscribedTopics map[string]*queueConsumer
 	simpleConsumers  map[string]*queueConsumer
 	closeOnce        sync.Once
+	sync.Mutex
 }
 
 const (
@@ -83,7 +85,7 @@ func (qc *queueConsumer) kill() {
 }
 
 func (qc *queueConsumer) startLoop(setRunning bool) {
-	logger.Printf("starting queue consumer %s", qc.queue.topic)
+	logger.Printf("starting queue consumer %s (set-running=%t)", qc.queue.topic, setRunning)
 	// not bound or already running
 	if qc.state.IsState(unbound) || qc.state.IsState(running) || qc.state.IsState(stopping) {
 		panic(fmt.Errorf("the queue consumer %s is in state %v. Cannot start", qc.queue.topic, qc.state.State()))
@@ -247,15 +249,19 @@ func (tc *consumer) Events() <-chan kafka.Event {
 // The consumerMock simply marks the topics as handled to make sure to
 // pass emitted messages back to the processor.
 func (tc *consumer) Subscribe(topics map[string]int64) error {
+	log.Printf("consumer: subscribing to topics: %v", topics)
+	var anyTopic string
 	for topic := range topics {
+		anyTopic = topic
 		if _, exists := tc.subscribedTopics[topic]; exists {
 			logger.Printf("consumer for %s already exists. This is strange", topic)
 		}
 		logger.Printf("Subscribe %s", topic)
 		tc.subscribedTopics[topic] = tc.tester.getOrCreateQueue(topic).bindConsumer(tc, true)
-		tc.subscribedTopics[topic].rebalance()
 		tc.subscribedTopics[topic].startLoop(false)
 	}
+
+	tc.subscribedTopics[anyTopic].rebalance()
 	return nil
 }
 
@@ -278,6 +284,8 @@ func (tc *consumer) Commit(topic string, partition int32, offset int64) error {
 // AddPartition marks the topic as a table topic.
 // The mock has to know the group table topic to ignore emit calls (which would never be consumed)
 func (tc *consumer) AddPartition(topic string, partition int32, initialOffset int64) error {
+	tc.Lock()
+	defer tc.Unlock()
 	logger.Printf("AddPartition %s", topic)
 	var firstStart bool
 	if _, exists := tc.simpleConsumers[topic]; !exists {

tester/queuetracker.go

@@ -57,7 +57,7 @@ func (mt *QueueTracker) Hwm() int64 {
 	return mt.tester.queueForTopic(mt.topic).hwm
 }
 
-// Hwm returns the tracker's next offset
+// NextOffset returns the tracker's next offset
 func (mt *QueueTracker) NextOffset() int64 {
 	return mt.nextOffset
 }

tester/tester.go

@@ -69,7 +69,7 @@ func (km *Tester) queueForTopic(topic string) *queue {
 	return q
 }
 
-// NewQueueTrackerFromEnd creates a message tracker that starts tracking
+// NewQueueTracker creates a message tracker that starts tracking
 // the messages from the end of the current queues
 func (km *Tester) NewQueueTracker(topic string) *QueueTracker {
 	km.waitStartup()