Offset Manager: report unknown topic errors

We should retry them still (if topic auto-create is enabled that is enough) but
we should also tell the user since if auto-create is *off* we otherwise just
appear to hang.

CHANGELOG.md

@@ -33,6 +33,8 @@ Bug Fixes:
    ([#685](https://github.com/Shopify/sarama/pull/685)).
  - Fix a possible tight loop in the consumer
    ([#693](https://github.com/Shopify/sarama/pull/693)).
+ - Report UnknownTopicOrPartition errors from the offset manager
+   ([#706](https://github.com/Shopify/sarama/pull/706)).
  - Fix possible negative partition value from the HashPartitioner
    ([#709](https://github.com/Shopify/sarama/pull/709)).
 

offset_manager.go

@@ -457,7 +457,7 @@ func (bom *brokerOffsetManager) flushToBroker() {
 		case ErrNoError:
 			block := request.blocks[s.topic][s.partition]
 			s.updateCommitted(block.offset, block.metadata)
-		case ErrUnknownTopicOrPartition, ErrNotLeaderForPartition, ErrLeaderNotAvailable,
+		case ErrNotLeaderForPartition, ErrLeaderNotAvailable,
 			ErrConsumerCoordinatorNotAvailable, ErrNotCoordinatorForConsumer:
 			// not a critical error, we just need to redispatch
 			delete(bom.subscriptions, s)
@@ -468,6 +468,12 @@ func (bom *brokerOffsetManager) flushToBroker() {
 		case ErrOffsetsLoadInProgress:
 			// nothing wrong but we didn't commit, we'll get it next time round
 			break
+		case ErrUnknownTopicOrPartition:
+			// let the user know *and* try redispatching - if topic-auto-create is
+			// enabled, redispatching should trigger a metadata request and create the
+			// topic; if not then re-dispatching won't help, but we've let the user
+			// know and it shouldn't hurt either (see https://github.com/Shopify/sarama/issues/706)
+			fallthrough
 		default:
 			// dunno, tell the user and try redispatching
 			s.handleError(err)