Can MockBroker return v2 messages?

[sanchezl]

Versions

Sarama Version: v1.12.0
Kafka Version: 0.10.1.0
Go Version: 1.8.3

Configuration

Kafka: using sarama.MockBroker
Sarama: Config.Version = sarama.V0_10_1_0

Logs

=== RUN   TestVersionedConsumer
=== RUN   TestVersionedConsumer/0.9.0.1
[sarama] 2017/08/24 12:36:29 *** mockbroker/0 listening on 127.0.0.1:54921
[sarama] 2017/08/24 12:36:29 Initializing new client
[sarama] 2017/08/24 12:36:29 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2017/08/24 12:36:29 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2017/08/24 12:36:29 client/metadata fetching metadata for all topics from broker 127.0.0.1:54921
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/0: connection opened
[sarama] 2017/08/24 12:36:29 Connected to broker at 127.0.0.1:54921 (unregistered)
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/0: served &{0 sarama 0xc42016a020} -> &{[0xc420172000] [0xc42016c060]}
[sarama] 2017/08/24 12:36:29 client/brokers registered new broker #0 at 127.0.0.1:54921
[sarama] 2017/08/24 12:36:29 Successfully initialized new client
[sarama] 2017/08/24 12:36:29 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2017/08/24 12:36:29 Connected to broker at 127.0.0.1:54921 (registered as #0)
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/1: connection opened
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/1: served &{0 sarama 0xc4201ae040} -> &{0 map[my_topic:map[0:0xc4201a6180]]}
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/1: served &{1 sarama 0xc4201702a0} -> &{0 map[my_topic:map[0:0xc42016c7e0]]}
[sarama] 2017/08/24 12:36:29 consumer/broker/0 added subscription to my_topic/0
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/1: served &{2 sarama 0xc4200f9480} -> &{map[my_topic:map[0:0xc4201cc030]] 0s 0}
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/1: invalid request: err=read tcp 127.0.0.1:54921->127.0.0.1:54923: use of closed network connection, (*sarama.request)(<nil>)
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/1: connection closed, err=<nil>
[sarama] 2017/08/24 12:36:29 consumer/broker/0 disconnecting due to error processing FetchRequest: EOF
[sarama] 2017/08/24 12:36:29 Closed connection to broker 127.0.0.1:54921
[sarama] 2017/08/24 12:36:29 kafka: error while consuming my_topic/0: EOF
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/0: invalid request: err=read tcp 127.0.0.1:54921->127.0.0.1:54922: use of closed network connection, (*sarama.request)(<nil>)
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/0: connection closed, err=<nil>
[sarama] 2017/08/24 12:36:29 *** mockbroker/0: listener closed, err=accept tcp 127.0.0.1:54921: use of closed network connection
=== RUN   TestVersionedConsumer/0.10.1.0
[sarama] 2017/08/24 12:36:29 *** mockbroker/0 listening on 127.0.0.1:54924
[sarama] 2017/08/24 12:36:29 Initializing new client
[sarama] 2017/08/24 12:36:29 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2017/08/24 12:36:29 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2017/08/24 12:36:29 client/metadata fetching metadata for all topics from broker 127.0.0.1:54924
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/0: connection opened
[sarama] 2017/08/24 12:36:29 Connected to broker at 127.0.0.1:54924 (unregistered)
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/0: served &{0 sarama 0xc4200f98a0} -> &{[0xc4200b6c60] [0xc4201cc810]}
[sarama] 2017/08/24 12:36:29 client/brokers registered new broker #0 at 127.0.0.1:54924
[sarama] 2017/08/24 12:36:29 Successfully initialized new client
[sarama] 2017/08/24 12:36:29 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/1: connection opened
[sarama] 2017/08/24 12:36:29 Connected to broker at 127.0.0.1:54924 (registered as #0)
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/1: served &{0 sarama 0xc42010a890} -> &{0 map[my_topic:map[0:0xc4201ccc00]]}
[sarama] 2017/08/24 12:36:29 Closed connection to broker 127.0.0.1:54924
[sarama] 2017/08/24 12:36:29 client/metadata fetching metadata for [my_topic] from broker 127.0.0.1:54924
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/1: invalid request: err=EOF, (*sarama.request)(<nil>)
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/1: connection closed, err=<nil>
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/0: served &{1 sarama 0xc4200f9b40} -> &{[0xc4200b6f20] [0xc4201cced0]}
[sarama] 2017/08/24 12:36:29 ClientID is the default of 'sarama', you should consider setting it to something application-specific.
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/2: connection opened
[sarama] 2017/08/24 12:36:29 Connected to broker at 127.0.0.1:54924 (registered as #0)
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/2: served &{1 sarama 0xc4201fe170} -> &{0 map[my_topic:map[0:0xc4201fc600]]}
[sarama] 2017/08/24 12:36:29 Closed connection to broker 127.0.0.1:54924
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/2: invalid request: err=EOF, (*sarama.request)(<nil>)
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/2: connection closed, err=<nil>
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/0: invalid request: err=read tcp 127.0.0.1:54924->127.0.0.1:54925: use of closed network connection, (*sarama.request)(<nil>)
[sarama] 2017/08/24 12:36:29 *** mockbroker/0/0: connection closed, err=<nil>
[sarama] 2017/08/24 12:36:29 *** mockbroker/0: listener closed, err=accept tcp 127.0.0.1:54924: use of closed network connection
--- FAIL: TestVersionedConsumer (0.01s)
    --- PASS: TestVersionedConsumer/0.9.0.1 (0.00s)
    --- FAIL: TestVersionedConsumer/0.10.1.0 (0.00s)
    	main_test.go:53: kafka: insufficient data to decode packet, more bytes expected
FAIL

Problem Description

My product's configuration defaults to using sarama.V0_9_0_1 as the kafka version when creating consumers and producers (we embed Kafka 0.9.0.1). I'm changing the embedded kafka to version 0.10.1.0, and as such, have updated the KafkaVersion to match. As such, when my unit tests run, which use MockBroker, I get the error: **

kafka: insufficient data to decode packet, more bytes expected** when trying to consume a message from the MockBroker.

Is this a bug (e.g. the Consumer should not require the Timestamp field to be set in the v2 message), or is there a way of making MockBroker publish a v2 FetchResponse (e.g. fill in the Timestamp field in it's response).

Sample test: Archive.zip

[eapache]

It's an oversight that you can't make the mock publish a v2 response, I'll add this shortly.