Better locking and logging.

consumergroup/consumer_group.go

@@ -243,26 +243,7 @@ func (cg *ConsumerGroup) CommitUpto(message *sarama.ConsumerMessage) error {
 	return nil
 }
 
-func (cg *ConsumerGroup) closeOnPanic() {
-	if err := recover(); err != nil {
-		cg.Logf("Error: %s\n", err)
-
-		// Try to produce an error event on the channel so we can inform the consumer.
-		// If that doesn't work, continue.
-		ce := &sarama.ConsumerError{Err: fmt.Errorf("%s", err)}
-		select {
-		case cg.errors <- ce:
-		default:
-		}
-
-		// Now, close the consumer
-		cg.Close()
-	}
-}
-
 func (cg *ConsumerGroup) topicListConsumer(topics []string) {
-	defer cg.closeOnPanic()
-
 	for {
 		select {
 		case <-cg.stopper:
@@ -272,7 +253,8 @@ func (cg *ConsumerGroup) topicListConsumer(topics []string) {
 
 		consumers, consumerChanges, err := cg.zk.Consumers(cg.name)
 		if err != nil {
-			panic(err)
+			cg.Logf("FAILED to get list of registered consumer instances: %s\n", err)
+			return
 		}
 
 		cg.consumers = consumers
@@ -299,7 +281,6 @@ func (cg *ConsumerGroup) topicListConsumer(topics []string) {
 }
 
 func (cg *ConsumerGroup) topicConsumer(topic string, messages chan<- *sarama.ConsumerMessage, errors chan<- *sarama.ConsumerError, stopper <-chan struct{}) {
-	defer cg.closeOnPanic()
 	defer cg.wg.Done()
 
 	select {
@@ -313,7 +294,8 @@ func (cg *ConsumerGroup) topicConsumer(topic string, messages chan<- *sarama.Con
 	// Fetch a list of partition IDs
 	partitions, err := cg.zk.Partitions(topic)
 	if err != nil {
-		panic(err)
+		cg.Logf("%s :: FAILED to get list of partitions: %s\n", topic, err)
+		return
 	}
 
 	dividedPartitions := dividePartitionsBetweenConsumers(cg.consumers, partitions)
@@ -334,7 +316,6 @@ func (cg *ConsumerGroup) topicConsumer(topic string, messages chan<- *sarama.Con
 
 // Consumes a partition
 func (cg *ConsumerGroup) partitionConsumer(topic string, partition int32, messages chan<- *sarama.ConsumerMessage, errors chan<- *sarama.ConsumerError, wg *sync.WaitGroup, stopper <-chan struct{}) {
-	defer cg.closeOnPanic()
 	defer wg.Done()
 
 	select {
@@ -345,13 +326,15 @@ func (cg *ConsumerGroup) partitionConsumer(topic string, partition int32, messag
 
 	err := cg.zk.Claim(cg.name, topic, partition, cg.id)
 	if err != nil {
-		panic(err)
+		cg.Logf("%s/%d :: FAILED to claim the partition: %s\n", topic, partition, err)
+		return
 	}
 	defer cg.zk.Release(cg.name, topic, partition, cg.id)
 
 	nextOffset, err := cg.offsetManager.InitializePartition(topic, partition)
 	if err != nil {
-		panic(err)
+		cg.Logf("%s/%d :: FAILED to determine initial offset: %s\n", topic, partition, err)
+		return
 	}
 
 	if nextOffset > 0 {
@@ -367,7 +350,8 @@ func (cg *ConsumerGroup) partitionConsumer(topic string, partition int32, messag
 
 	consumer, err := cg.consumer.ConsumePartition(topic, partition, nextOffset)
 	if err != nil {
-		panic(err)
+		cg.Logf("%s/%d :: FAILED to start partition consumer: %s\n", topic, partition, err)
+		return
 	}
 	defer consumer.Close()
 

consumergroup/offset_manager.go

@@ -42,7 +42,7 @@ type OffsetManager interface {
 }
 
 var (
-	UncleanClose = errors.New("Not all offsets were committed before shutdown was completed.")
+	UncleanClose = errors.New("Not all offsets were committed before shutdown was completed")
 )
 
 // OffsetManagerConfig holds configuration setting son how the offset manager should behave.
@@ -124,10 +124,10 @@ func (zom *zookeeperOffsetManager) InitializePartition(topic string, partition i
 }
 
 func (zom *zookeeperOffsetManager) FinalizePartition(topic string, partition int32, lastOffset int64, timeout time.Duration) error {
-	zom.l.Lock()
-	defer zom.l.Unlock()
-
+	zom.l.RLock()
 	tracker := zom.offsets[topic][partition]
+	zom.l.RUnlock()
+
 	if !tracker.waitForOffset(lastOffset, timeout) {
 		return fmt.Errorf("TIMEOUT waiting for offset %d. Last committed offset: %d", lastOffset, tracker.lastCommittedOffset)
 	}
@@ -136,7 +136,10 @@ func (zom *zookeeperOffsetManager) FinalizePartition(topic string, partition int
 		return fmt.Errorf("FAILED to commit offset %d to Zookeeper. Last committed offset: %d", tracker.highestProcessedOffset, tracker.lastCommittedOffset)
 	}
 
+	zom.l.Lock()
 	delete(zom.offsets[topic], partition)
+	zom.l.Unlock()
+
 	return nil
 }
 
@@ -154,10 +157,9 @@ func (zom *zookeeperOffsetManager) Close() error {
 	defer zom.l.Unlock()
 
 	var closeError error
-	for topic, partitionOffsets := range zom.offsets {
+	for _, partitionOffsets := range zom.offsets {
 		if len(partitionOffsets) > 0 {
 			closeError = UncleanClose
-			zom.cg.Logf("WARNING: uncommitted offsets for topic %s.\n", topic)
 		}
 	}