VOL-1460 Updated core to use voltha-protos instead of local protos
Moved protos to python directory in order to maintain functionaly of containers built there.
Added capability to do local builds of protos
Added instructions on running dep ensure for getting protos.
Updated github.com/golang/protobuf version to v1.3.1
Change-Id: Ia6ef55f07f0d5dcb5b750d7c37b21b71db85bfc4
diff --git a/vendor/gopkg.in/Shopify/sarama.v1/consumer.go b/vendor/gopkg.in/Shopify/sarama.v1/consumer.go
index 33d9d14..ce72ff1 100644
--- a/vendor/gopkg.in/Shopify/sarama.v1/consumer.go
+++ b/vendor/gopkg.in/Shopify/sarama.v1/consumer.go
@@ -314,6 +314,8 @@
fetchSize int32
offset int64
+
+ retries int32
}
var errTimedOut = errors.New("timed out feeding messages to the user") // not user-facing
@@ -332,12 +334,21 @@
}
}
+func (child *partitionConsumer) computeBackoff() time.Duration {
+ if child.conf.Consumer.Retry.BackoffFunc != nil {
+ retries := atomic.AddInt32(&child.retries, 1)
+ return child.conf.Consumer.Retry.BackoffFunc(int(retries))
+ } else {
+ return child.conf.Consumer.Retry.Backoff
+ }
+}
+
func (child *partitionConsumer) dispatcher() {
for range child.trigger {
select {
case <-child.dying:
close(child.trigger)
- case <-time.After(child.conf.Consumer.Retry.Backoff):
+ case <-time.After(child.computeBackoff()):
if child.broker != nil {
child.consumer.unrefBrokerConsumer(child.broker)
child.broker = nil
@@ -451,6 +462,10 @@
for response := range child.feeder {
msgs, child.responseResult = child.parseResponse(response)
+ if child.responseResult == nil {
+ atomic.StoreInt32(&child.retries, 0)
+ }
+
for i, msg := range msgs {
messageSelect:
select {
@@ -487,9 +502,13 @@
for _, msgBlock := range msgSet.Messages {
for _, msg := range msgBlock.Messages() {
offset := msg.Offset
+ timestamp := msg.Msg.Timestamp
if msg.Msg.Version >= 1 {
baseOffset := msgBlock.Offset - msgBlock.Messages()[len(msgBlock.Messages())-1].Offset
offset += baseOffset
+ if msg.Msg.LogAppendTime {
+ timestamp = msgBlock.Msg.Timestamp
+ }
}
if offset < child.offset {
continue
@@ -500,14 +519,14 @@
Key: msg.Msg.Key,
Value: msg.Msg.Value,
Offset: offset,
- Timestamp: msg.Msg.Timestamp,
+ Timestamp: timestamp,
BlockTimestamp: msgBlock.Msg.Timestamp,
})
child.offset = offset + 1
}
}
if len(messages) == 0 {
- return nil, ErrIncompleteResponse
+ child.offset++
}
return messages, nil
}
@@ -519,19 +538,23 @@
if offset < child.offset {
continue
}
+ timestamp := batch.FirstTimestamp.Add(rec.TimestampDelta)
+ if batch.LogAppendTime {
+ timestamp = batch.MaxTimestamp
+ }
messages = append(messages, &ConsumerMessage{
Topic: child.topic,
Partition: child.partition,
Key: rec.Key,
Value: rec.Value,
Offset: offset,
- Timestamp: batch.FirstTimestamp.Add(rec.TimestampDelta),
+ Timestamp: timestamp,
Headers: rec.Headers,
})
child.offset = offset + 1
}
if len(messages) == 0 {
- child.offset += 1
+ child.offset++
}
return messages, nil
}
@@ -787,6 +810,9 @@
MinBytes: bc.consumer.conf.Consumer.Fetch.Min,
MaxWaitTime: int32(bc.consumer.conf.Consumer.MaxWaitTime / time.Millisecond),
}
+ if bc.consumer.conf.Version.IsAtLeast(V0_9_0_0) {
+ request.Version = 1
+ }
if bc.consumer.conf.Version.IsAtLeast(V0_10_0_0) {
request.Version = 2
}