Update producer and consumer to allow sending and receiving Key/Value records

[nicholastmosher]

Here's a list of changes in this PR:

• Added Record::key(&self) -> &[u8] and Record::value(&self) -> &[u8] in the consumer
• Removed impl AsRef<[u8]> for Record in consumer because we now want to explicitly distinguish between keys and values
• Added TopicProducer::send_key_value_record(&self, key: K, value: V, partition: i32) to producer
• Changed the representation of record keys in the protocol from key: B to key: Option<B> (where B is a byte buffer)
• Added a test that shows that decoding a record with an old encoding just so happens to work as expected
  • Old records are always represented as "empty keys" in the form of zero-length buffers
  • Zero-length buffers are encoded as a length whose value is zero 0x00.
  • The new record encoding represents "absent keys" in the form of Option::None
  • Option::None tag is encoded as 0x00 (as opposed to Option::Some, whose tag is 0x01)
  • Therefore, decoding old records with empty keys into new records with absent keys works out-of-the-box
  • However, we must never attempt to decode a new record with any key using the older decoder.

Some potentially-open-questions:

• Are we happy with the new function names? These names are all potentially up for debate:
  • Record::key
  • Record::value
  • TopicProducer::send_key_value_record
  • TopicProducer::send

[ajhunyady]

@nicholastmosher, is there a sample CLI on how to use this feature? We need to prepare a simple app or blog to introduce it.

[ajhunyady]

Hmmm... this looks very complicated to me and it shouldn't be.

[sehz]

For CLI, here is one way to do:

$ fluvio produce --key-seperator="="
name=John Doe 
country=US

[sehz]

We should simplify API much as possible.
In most cases, K/V will be typical:

producer.send("name","doe")

For case, where only values is send:

producer.send(None,"doe")

[ajhunyady]

I suggest keeping the .send() for values and have an alternative API for send KV. That way the most common use case send will remain unchanged.

[sehz]

K/V will be more typical use case rather than other way around.

[ajhunyady]

In that case, send it is...

[ajhunyady]

For CLI, here is one way to do:

$ fluvio produce --key-seperator="="
name=John Doe 
country=US

There should be some indication this is a key/value record, I assume is `--key-separator
I would use an example that's more realistic

$ fluvio produce --key-separator="="
cust_100="John Doe, Los Angeles, CA, USA" 

[sehz]

--kv to indicate kv value.

$ fluvio produce --kv --separator "="
name=John Doe 
country=US

[sehz]

@simlay Can you review this as well?

[simlay]

Sehyo managed to cover most of my concerns about things. I had some minor nits.

Also what's the file src/dataplane-protocol/tests/test_old_record_empty_key.bin?

[nicholastmosher]

@simlay that .bin file is what I used to verify that records in the new schema can be successfully decoded from old records that were encoded with the old schema. See https://github.com/infinyon/fluvio/pull/828/files#diff-e50536c96490b842b8f91fab3fca42353b0beca28ebe99ecd1eb80bcfc53c4b6R568-R606

Basically, I used the old encoder to encode a specific record into bytes, and saved those bytes into test_old_record_empty_key.bin. Then I wrote a test which attempts to decode a new-schema-record from those same bytes, and assert that the data landed where it was supposed to in the new record.

[simlay]

I think I'm good with this.

[sehz]

Can you squash commits and produce clean commit logs?