This is a notation for handling repeated structures. These will always be encoded as an int32 size containing the length N followed by N repetitions of the structure which can itself be made up of other primitive types. In the BNF grammars below we will show an array of a structure foo as [foo].
The BNFs below give an exact context free grammar for the request and response binary format. For each API I will give the request and response together followed by all the sub-definitions. The BNF is intentionally not compact in order to give human-readable name (for example I define a production for ErrorCode even though it is just an int16 in order to give it a symbolic name). As always in a BNF a sequence of productions indicates concatenation, so the MetadataRequest given below would be a sequence of bytes containing first a VersionId, then a ClientId, and then an array of TopicNames (each of which has its own definition). Productions are always given in camel case and primitive types in lower case. When there are multiple possible productions these are separated with '|' and may be enclosed in parenthesis for grouping. The top-level definition is always given first and subsequent sub-parts are indented.
-All requests and responses originate from the following grammar which will be incrementally describe through the rest of this document:
The response will always match the paired request (e.g. we will send a MetadataResponse in return to a MetadataRequest).
One structure common to both the produce and fetch requests is the message set format. A message in kafka is a key-value pair with a small amount of associated metadata. A message set is just a sequence of messages with offset and size information. This format happens to be used both for the on-disk storage on the broker and the on-the-wire format.
+One structure model to both the produce and fetch requests is the message set format. A message in kafka is a key-value pair with a small amount of associated metadata. A message set is just a sequence of messages with offset and size information. This format happens to be used both for the on-disk storage on the broker and the on-the-wire format.
A message set is also the unit of compression in Kafka, and we allow messages to recursively contain compressed message sets to allow batch compression.
N.B., MessageSets are not preceded by an int32 like other array elements in the protocol.
Some people have asked why we don't use HTTP. There are a number of reasons, the best is that client implementors can make use of some of the more advanced TCP features--the ability to multiplex requests, the ability to simultaneously poll many connections, etc. We have also found HTTP libraries in many languages to be surprisingly shabby.
Others have asked if maybe we shouldn't support many different protocols. Prior experience with this was that it makes it very hard to add and test new features if they have to be ported across many protocol implementations. Our feeling is that most users don't really see multiple protocols as a feature, they just want a good reliable client in the language of their choice.
Another question is why we don't adopt XMPP, STOMP, AMQP or an existing protocol. The answer to this varies by protocol, but in general the problem is that the protocol does determine large parts of the implementation and we couldn't do what we are doing if we didn't have control over the protocol. Our belief is that it is possible to do better than existing messaging systems have in providing a truly distributed messaging system, and to do this we need to build something that works differently.
diff --git a/tools/kafpro/command.go b/tools/kafpro/command.go index 353db56..be200ab 100644 --- a/tools/kafpro/command.go +++ b/tools/kafpro/command.go @@ -11,8 +11,8 @@ import ( "sync/atomic" "time" - "h12.me/kafka/cluster" "h12.me/kafka/consumer" + "h12.me/kafka/model" "h12.me/kafka/proto" ) @@ -114,7 +114,7 @@ type ConsumeCommand struct { Count bool `long:"count"` } -func (cmd *ConsumeCommand) Exec(cl *cluster.C) error { +func (cmd *ConsumeCommand) Exec(cl model.Cluster) error { // TODO: detect format partitions, err := cl.Partitions(cmd.Topic) if err != nil { @@ -203,7 +203,7 @@ type CommitCommand struct { Retention int `long:"retention"` // millisecond } -func (cmd *CommitCommand) Exec(cl *cluster.C) error { +func (cmd *CommitCommand) Exec(cl model.Cluster) error { /* req := &broker.Request{ ClientID: clientID, diff --git a/tools/kafpro/main.go b/tools/kafpro/main.go index 6622ee3..09cbeb4 100644 --- a/tools/kafpro/main.go +++ b/tools/kafpro/main.go @@ -44,10 +44,7 @@ func main() { if err != nil { log.Fatal(err) } - c, err := cluster.New(broker.New, cfg.Brokers) - if err != nil { - log.Fatal(err) - } + c := cluster.New(broker.New, cfg.Brokers) //fmt.Println(toJSON(cfg)) switch parser.Active.Name { case "consume":