fixes according to comments

Author: replay

kafka/consumer.go

@@ -303,12 +303,12 @@ func (c *Consumer) startConsumer() error {
 			var currentOffset int64
 			switch c.conf.StartAtOffset {
 			case "oldest":
-				currentOffset, _, err = c.tryGetOffset(topic, partition, int64(confluent.OffsetBeginning), 3, time.Second)
+				currentOffset, err = c.tryGetOffset(topic, partition, int64(confluent.OffsetBeginning), 3, time.Second)
 				if err != nil {
 					return err
 				}
 			case "newest":
-				_, currentOffset, err = c.tryGetOffset(topic, partition, int64(confluent.OffsetEnd), 3, time.Second)
+				currentOffset, err = c.tryGetOffset(topic, partition, int64(confluent.OffsetEnd), 3, time.Second)
 				if err != nil {
 					return err
 				}
@@ -318,9 +318,13 @@ func (c *Consumer) startConsumer() error {
 					return fmt.Errorf("invalid offest format %s: %s", c.conf.StartAtOffset, err)
 				}
 				currentOffset = time.Now().Add(-1*offsetDuration).UnixNano() / int64(time.Millisecond)
-				currentOffset, _, err = c.tryGetOffset(topic, partition, currentOffset, 3, time.Second)
+				currentOffset, err = c.tryGetOffset(topic, partition, currentOffset, 3, time.Second)
 				if err != nil {
-					return err
+					log.Warn("kafka-consumer: Failed to get specified offset %s, falling back to \"oldest\"", c.conf.StartAtOffset)
+					currentOffset, err = c.tryGetOffset(topic, partition, int64(confluent.OffsetBeginning), 3, time.Second)
+					if err != nil {
+						return err
+					}
 				}
 			}
 
@@ -344,44 +348,50 @@ func (c *Consumer) startConsumer() error {
 	return c.consumer.Assign(topicPartitions)
 }
 
-func (c *Consumer) tryGetOffset(topic string, partition int32, offsetI int64, attempts int, sleep time.Duration) (int64, int64, error) {
+func (c *Consumer) tryGetOffset(topic string, partition int32, offsetI int64, attempts int, sleep time.Duration) (int64, error) {
 	offset, err := confluent.NewOffset(offsetI)
 	if err != nil {
-		return 0, 0, err
+		return 0, err
 	}
 
-	var val1, val2 int64
+	var beginning, end int64
 
 	attempt := 1
 	for {
 		if offset == confluent.OffsetBeginning || offset == confluent.OffsetEnd {
-			val1, val2, err = c.consumer.QueryWatermarkOffsets(topic, partition, c.conf.MetadataTimeout)
+			beginning, end, err = c.consumer.QueryWatermarkOffsets(topic, partition, c.conf.MetadataTimeout)
+			if err == nil {
+				if offset == confluent.OffsetBeginning {
+					return beginning, nil
+				} else {
+					return end, nil
+				}
+			}
 		} else {
 			times := []confluent.TopicPartition{{Topic: &topic, Partition: partition, Offset: offset}}
 			times, err = c.consumer.OffsetsForTimes(times, c.conf.MetadataTimeout)
-			if err == nil {
-				if len(times) == 0 {
-					err = fmt.Errorf("Got 0 topics returned from broker")
+			if err != nil || len(times) == 0 {
+				if err == nil {
+					err = fmt.Errorf("Failed to get offset %d from kafka, falling back to \"oldest\"", offset)
 				} else {
-					val1 = int64(times[0].Offset)
+					err = fmt.Errorf("Failed to get offset %d from kafka, falling back to \"oldest\": %s", offset, err)
 				}
+				offset = confluent.OffsetBeginning
+			} else {
+				return int64(times[0].Offset), nil
 			}
 		}
 
-		if err == nil {
-			return val1, val2, err
-		}
-
 		if attempt >= attempts {
 			break
 		}
 
-		log.Warn("kafka-consumer: Error when qerying offsets, %d retries left: %s", attempts-attempt, err)
+		log.Warn("kafka-consumer: Error when querying offsets, %d retries left: %s", attempts-attempt, err)
 		attempt += 1
 		time.Sleep(sleep)
 	}
 
-	return 0, 0, fmt.Errorf("Failed to get offset %s of partition %s:%d. %s (attempt %d/%d)", offset.String(), topic, partition, err, attempt, attempts)
+	return 0, fmt.Errorf("Failed to get offset %s of partition %s:%d. %s (attempt %d/%d)", offset.String(), topic, partition, err, attempt, attempts)
 }
 
 func (c *Consumer) Stop() {