Fix offset-manager expiry timestamps to 'now'

Thanks to Maxim for pointing out that 0 doesn't mean now, it means unix epoch 0
(a long time ago).

Fixes #530.

offset_manager.go

@@ -476,7 +476,7 @@ func (bom *brokerOffsetManager) constructRequest() *OffsetCommitRequest {
 	for s := range bom.subscriptions {
 		s.lock.Lock()
 		if s.dirty {
-			r.AddBlock(s.topic, s.partition, s.offset, 0, s.metadata)
+			r.AddBlock(s.topic, s.partition, s.offset, ReceiveTime, s.metadata)
 		}
 		s.lock.Unlock()
 	}