Fix race in broker version check

[eapache]

Move it until after we're sure we have a version to check against. Fixes a race
found by @kchaliki (#678 (comment)).

Also make the which-version-do-I-need part of the protocol interface where it
belongs.

[kchaliki]

@eapache thanks for the fix, I can confirm I no longer get a panic due to the nil pointer, however I still don't see a timestamp applied to my messages - does it work for you?

[eapache]

I have not been able to properly test this yet; it seems there are some broker-side settings that may need to be set in addition to just producing with the right message version?

[kchaliki]

Indeed, I will dig some more

[kchaliki]

Confirmed after running

kafka-topics.sh --zookeeper blah --alter --topic mytopic --config message.timestamp.type=LogAppendTime

I get timestamps. I think one small issue is they are millisecond timestamps and that is not accounted for correctly when parsing the response and calling time.Unix(block.Timestamp, 0)

[kchaliki]

Something like this does the trick

timestampNsec := block.Timestamp % 1000
timestampSec := block.Timestamp / 1000
timestamp := time.Unix(timestampSec, timestampNsec)