Just some minor comments.
1. Struct is not a type. So, it probably should be in the protocol package,
instead of protocol.types.
2. Is the comment in Struct.instance() outdated? It seems this is only used
for creating an instance of a Field of type struct.
3. When creating a Field in a schema, it's a bit weird that we first create
a field using
Field(String name, Type type)
only to later recreate it in Schema using
Field(int index, String name, Type type, String doc, Object defaultValue,
Schema schema)
An alternative way is sth like the following:
new Schema().addField("f1", Int).addField("f2", Long)
This would avoid the double creation of Field. Not sure if it's better.
4. With this approach, refactoring field names will be harder since there
won't be ide support.
Thanks,
Jun
On Fri, Feb 7, 2014 at 12:56 PM, Jay Kreps <jay.kr...@gmail.com> wrote:
> Okay this is the last discussion item for the new client code. :-)
>
> Previously to define an api you would implement a request and response
> scala object that read and wrote its own bytes. There were a few problems
> with this:
> 1. The consistency of the protocol was very hard to maintain.
> 2. You ended up hand-coding size estimation which was very tedious and
> error prone
> 3. Error messages wouldn't give any field name information you would just
> get some kind of BufferUnderflowException with no information about what or
> why. Fixing these were hard because each object would have to implement
> this good error handling.
> 4. There wasn't good support for api versioning. We have an api version
> that is included in the request, but there was no easy way to maintain both
> the old format and the new format.
> 5. The header information was baked into each request and it was only
> though great care that we could keep the header standard throughout the
> requests.
> 6. The same class that defined the protocol was used throughout the code.
> So what were intended to be dumb DTOs ended up getting loaded up with
> domain logic. Invariably aspects of this representation would end up
> leaking into the protocol.
> 7. It was very hard to figure out what the protocol was from the code since
> the definition was embedded in byte munging code spread out over dozens of
> files.
>
> So that was definitely bad.
>
> We considered moving to an off-the-shelf protocol definition language like
> avro or protocol buffers. But prior experience with these is that they are
> great for whipping together a quick service but for a stable protocol it is
> actually better to define the protocol rather than specifying an
> implementation like avro or protocol buffers. This is similar to what is
> done with AMQP which I think does a fantastic job of providing a well
> specified messaging protocol (that protocol is not suitable for the type of
> system we are building, but their method of specifying it I think is very
> good).
>
> So the conclusion was to retain our BNF-specified protocol and instead
> implement a simple library for implementing this protocol. This would have
> the advantage of letting us retain our existing protocol and also to add a
> few Kafka-specific optimizations. This library is just a helper utility for
> implementing our protocol spec, the spec remains the source of truth.
>
> I implemented this as part of the new client effort. I will describe how my
> library works and the pattern I think we should use with it.
>
> The code for defining the protocol is in
> org.apache.kafka.common.protocol.types. Note that this is meant to be a
> stand-alone library for serialization, it doesn't know anything about our
> actual request and responses or even that the messages being defined will
> be sent over a network. The definition of our protocol is defined in
> org.apache.kafka.common.protocol.Protocol, this is just the protocol and is
> decoupled from the network layer and everything else.
>
> We define a set of types that match our protocol, namely:
> - fixed length primitives: int8, int16, int32, int64
> - variable-length primitives: string, bytes
> - container types: arrayof, struct
>
> You define a message using types. All types extend
> org.apache.kafka.common.protocol.types.Type.java. Each type knows how to
> read, write, validate, and estimate the size of a single java object type.
> Here is the correspondence
> Type.INT8: java.lang.Byte
> Type.INT16: java.lang.Short
> Type.INT32: java.lang.Integer
> Type.INT32: java.lang.Long
> Type.STRING: java.lang.String
> Type.BYTES: java.nio.ByteBuffer
> ArrayOf: Object[]
> Schema: Struct
> The correspondence here can be thought of as that between a class and an
> object: the class specifies the layout of the object, the object is an
> instantiation of that class with particular values. Each message is defined
> by a Schema, which can be used to read and write a Struct. The schema
> specifies the fields in the type, and the Struct is an "instantiation" of
> those fields with actual values. A struct can be thought of as a special
> purpose hashmap.
>
> An example will make this more clear. Here is how you define the request
> header schema:
> new Schema(new Field("api_key", INT16, "The id of the request type."),
> new Field("api_version", INT16, "The version of the API."),
> new Field("correlation_id", INT32, "documentation string"),
> new Field("client_id", STRING, "more documentation."));
>
> So a request header is a message that consists of a short api key followed
> by a short api version followed by a correlation id and client id.
>
> Here is a more complex example, the producer response:
>
> new Schema(new Field("responses", new ArrayOf(new Schema(new Field("topic",
> STRING), new Field("partition_responses", new ArrayOf(new Schema(new Field(
> "partition", INT32), new Field("error_code", INT16), new
> Field("base_offset"
> , INT64))))))))
>
> (indentation in email is tricky). Note that this has a schema which
> contains an array of sub-records which in turn have a sub-array of records.
> As this nesting gets more complicated it can get a bit hard to read, so you
> can break it up using variables. An equivalent definition would be:
>
> Schema partitionResponse = new Schema(new Field("partition", INT32),
>
> new Field("error_code", INT16),
>
> new Field("base_offset", INT64));
>
> Schema topicResponse = new Schema(new Field("topic", STRING),
>
> new Field("partition_responses", new
> ArrayOf(partitionResponse)));
>
> Schema producerResposne = new Schema(new Field("responses", new
> ArrayOf(topicResponse)));
>
> Note that this is exactly equivalent.
>
> Okay once such a schema is defined you can write an object in the following
> way:
>
> Struct header = new Struct(headerSchema);
>
> header.set("api_key", (short) 1);
>
> header.set("api_version", (short), 0);
>
> ...
>
> headerSchema.write(buffer, header);
>
> And you can read an instance of a header by doing:
>
> Struct header = headerSchema.read(buffer);
>
> Short apiKey = (Short) header.get("api_key");
>
> Field apiVersionField = header.field("api_version");
>
> Short apiKey = header.get(apiVersionField);
>
> Note the two different field access styles. Accessing a field by name has
> the performance of a hash table lookup. However for performance critical
> situations you can get the Field object that represents that entry in the
> struct. Getting this field object takes a hash table lookup but once you
> have it it will get that field out of any instance of that struct with the
> performance of an array access. So this is useful in cases where you can
> statically fetch all the fields and then use them on every request (and
> assuming you need to optimize performance).
>
> These raw structs are logic-free and act as the "DTO" for data that will be
> sent over the network.
>
> For the more complex requests and responses interacting with the raw struct
> is not very pleasent. My recommendation is that we still maintain a java
> object that is the "domain object" for the request and knows how to read
> and write itself to the struct. This is what you would end up passing down
> into KafkaApis. This will have all the convenience methods that people were
> wanting to add to the protocol objects before. The downside of this is that
> in some ways you define the request twice, but I think both of these layers
> are actually needed and would evolve independently (the struct only when
> the protocol changes and the domain object with the needs of the code that
> use it). I haven't actually done this in the produce yet, in part because I
> think to make these domain objects properly you need to use them on the
> server side too which we aren't ready for yet. However I did add a version
> of this for metadata on KAFKA-1238 here:
>
>
> https://issues.apache.org/jira/secure/attachment/12627654/KAFKA-1238-v1.patch
>
> Okay, it would be great to get feedback on this code and this general
> approach to protocol definition. If everyone likes it then I am going to
> consider all the discussion items for the new code wrapped up and move on
> to the more detailed code review and testing.
>
> -Jay
>
