{"action":"created","comment":{"url":"https://api.github.com/repos/klarna/kafka_protocol/pulls/comments/185427815","pull_request_review_id":116812116,"id":185427815,"diff_hunk":"@@ -138,764 +213,302 @@\n                 | struct_schema()\n                 | {array, schema()}\n                 | decode_fun(). %% caller defined decoder\n--type stack() :: [{tag(), vsn()} | field_name()]. %% encode / decode stack\n-\n--define(INT, signed-integer).\n--define(SCHEMA_MODULE, kpro_schema).\n--define(PRELUDE, kpro_prelude_schema).\n--define(NO_TIMESTAMP, -1).\n+-type stack() :: [{api(), vsn()} | field_name()]. %% encode / decode stack\n+-type isolation_level() :: read_committed | read_uncommitted.\n+-type connection() :: kpro_connection:connection().\n+-type conn_config() :: kpro_connection:config().\n+-type vsn_range() :: {vsn(), vsn()}.\n+-type vsn_ranges() :: #{api() => vsn_range()}.\n+-type protocol() :: plaintext | ssl | sasl_plaintext | sasl_ssl.\n+-type coordinator_type() :: group | txn.\n+-type group_id() :: binary().\n+-type transactional_id() :: binary().\n+-type producer_epoch() :: int16().\n+-type txn_ctx() :: #{ connection => connection()\n+                    , transactional_id => transactional_id()\n+                    , producer_id => producer_id()\n+                    , producer_epoch => producer_id()\n+                    }.\n+-type batch_meta_key() ::\n+        is_transaction % if this message was produced in a transaction\n+      | is_control % for read_uncommitted clients\n+      | last_offset % client wont have to do lists:last(Messages)\n+      | max_ts % client don't have to scan the messages for max\n+      | producer_id. % it can be referenced by a future fetch response in\n+                     % its aborted_transactions field\n+-type batch_meta_val() :: boolean() | atom() | integer().\n+-type batch_meta() :: ?KPRO_NO_BATCH_META %% magic 0-1\n+                    | #{batch_meta_key() => batch_meta_val()}.\n+-type batch_decode_result() :: ?incomplete_batch(int32())\n+                             | {batch_meta(), [message()]}.\n+%% offset or offset + associated user-data to commit\n+-type offset_ud() :: offset() %% no user data\n+                   | {offset(), binary()}.\n+-type offsets_to_commit() :: #{{topic(), partition()} => offset_ud()}\n+                           | [{{topic(), partition()}, offset_ud()}].\n+\n+%% All versions of kafka messages (records) share the same header:\n+%% Offset => int64\n+%% Length => int32\n+%% We need to at least fetch 12 bytes in order to fetch:\n+%%  - one complete message when it's magic v0-1 not compressed\n+%%  - one comprete batch when it's v0-1 compressed batch\n+%%    v0-1 compressed batch is embedded in a wrapper message (i.e. recursive)\n+%%  - one complete batch when it is v2.\n+%%    v2 batch is flat and trailing the batch header.\n+-define(BATCH_LEADING_BYTES, 12).\n \n %%%_* APIs =====================================================================\n \n-%% @doc Return the allowed maximum correlation ID.\n--spec max_corr_id() -> corr_id().\n-max_corr_id() -> ?MAX_CORR_ID.\n-\n-%% @doc Help function to contruct a OffsetsRequest\n-%% against one single topic-partition.\n-%% @end\n--spec offsets_request(vsn(), topic(), partition(), msg_ts()) -> req().\n-offsets_request(Vsn, Topic, Partition, Time) ->\n-  PartitionFields =\n-    case Vsn of\n-      0 ->\n-        [{partition, Partition},\n-         {timestamp, Time},\n-         {max_num_offsets, 1}];\n-      1 ->\n-        [{partition, Partition},\n-         {timestamp, Time}]\n-    end,\n-  Fields =\n-    [{replica_id, ?KPRO_REPLICA_ID},\n-     {topics, [ [{topic, Topic},\n-                 {partitions, [ PartitionFields ]}]\n-              ]}\n-    ],\n-  req(offsets_request, Vsn, Fields).\n-\n-%% @doc Help function to construct a FetchRequest\n-%% against one single topic-partition.\n-%% @end\n--spec fetch_request(vsn(), topic(), partition(), offset(),\n-                    wait(), count(), count()) -> req().\n-fetch_request(Vsn, Topic, Partition, Offset,\n-              MaxWaitTime, MinBytes, MaxBytes) ->\n-  Fields0 =\n-    [{replica_id, ?KPRO_REPLICA_ID},\n-     {max_wait_time, MaxWaitTime},\n-     {min_bytes, MinBytes},\n-     {topics,[[{topic, Topic},\n-               {partitions,\n-                [[{partition, Partition},\n-                  {fetch_offset, Offset},\n-                  {max_bytes, MaxBytes}]]}]]}],\n-  %% Version 3 introduced a top level max_bytes\n-  %% we use the same value as per-partition max_bytes\n-  %% because this API is to build request against single partition\n-  Fields =\n-    case Vsn >= 3 of\n-      true  -> [{max_bytes, MaxBytes} | Fields0];\n-      false -> Fields0\n-    end,\n-  req(fetch_request, Vsn, Fields).\n-\n-%% @equiv produce_request(Vsn, Topic, Partition, KvList, RequiredAcks,\n-%%                        AckTimeout, no_compression)\n-%% @end\n--spec produce_request(vsn(), topic(), partition(), kv_list(),\n-                      required_acks(), wait()) -> req().\n-produce_request(Vsn, Topic, Partition, KvList, RequiredAcks, AckTimeout) ->\n-  produce_request(Vsn, Topic, Partition, KvList, RequiredAcks, AckTimeout,\n-                  no_compression).\n-\n-%% @doc Help function to construct a produce request for\n-%% messages targeting one single topic-partition.\n-%% @end\n--spec produce_request(vsn(), topic(), partition(), kv_list(),\n-                      required_acks(), wait(), compress_option()) -> req().\n-produce_request(Vsn, Topic, Partition, KvList,\n-                RequiredAcks, AckTimeout, CompressOption) ->\n-  Messages = encode_messages(KvList, CompressOption),\n-  Fields =\n-    [{acks, RequiredAcks},\n-     {timeout, AckTimeout},\n-     {topic_data, [[{topic, Topic},\n-                    {data, [[{partition, Partition},\n-                             {record_set, Messages}\n-                            ]]}\n-                   ]]}\n-    ],\n-  Req = req(produce_request, Vsn, Fields),\n-  Req#kpro_req{no_ack = RequiredAcks =:= 0}.\n-\n-%% @doc Help function to make a request body.\n--spec req(req_tag(), vsn(), struct()) -> req().\n-req(Tag, Vsn, Fields) ->\n-  #kpro_req{ tag = Tag\n-           , vsn = Vsn\n-           , msg = encode_struct(Tag, Vsn, Fields)\n-           }.\n-\n-%% @doc Get the next correlation ID.\n--spec next_corr_id(corr_id()) -> corr_id().\n-next_corr_id(?MAX_CORR_ID) -> 0;\n-next_corr_id(CorrId)       -> CorrId + 1.\n-\n-%% @doc Encode a request to bytes that can be sent on wire.\n+%% @doc Parse comma separated endpoints in a string into a list of\n+%% `{Host::string(), Port::integer()}' pairs.\n+parse_endpoints(String) ->\n+  parse_endpoints(undefined, String).\n+\n+%% @doc Same return value as `parse_endpoints/1'.\n+%% Endpoints may or may not start with protocol prefix (non case sensitive):\n+%% `PLAINTEXT://', `SSL://', `SASL_PLAINTEXT://' or `SASL_SSL://'.\n+%% The first arg is to filter desired endpoints from parse result.\n+-spec parse_endpoints(protocol() | undefined, string()) -> [endpoint()].\n+parse_endpoints(Protocol, String) ->\n+  kpro_lib:parse_endpoints(Protocol, String).\n+\n+%% @doc Help function to make a request. See also kpro_req_lib for more help\n+%% functions.\n+-spec make_request(api(), vsn(), struct()) -> req().\n+make_request(Api, Vsn, Fields) ->\n+  kpro_req_lib:make(Api, Vsn, Fields).\n+\n+%% @doc Help function to translate `#kpro_rsp{}' into a simpler term.\n+%% NOTE: This function is implemented with assumptions like:\n+%% topic-partition requests (list_offsets, produce, fetch etc.) are always sent\n+%% against only ONE topic-partition.\n+-spec parse_response(rsp()) -> term().\n+parse_response(Rsp) -> kpro_rsp_lib:parse(Rsp).\n+\n+%% @doc Encode request to byte stream.\n -spec encode_request(client_id(), corr_id(), req()) -> iodata().\n-encode_request(ClientId, CorrId0, Req) ->\n-  #kpro_req{tag = Tag, vsn = Vsn, msg = Msg} = Req,\n-  ApiKey = ?REQ_TO_API_KEY(Tag),\n-  true = (CorrId0 =< ?MAX_CORR_ID), %% assert\n-  true = (ApiKey < 1 bsl ?API_KEY_BITS), %% assert\n-  true = (Vsn < 1 bsl ?API_VERSION_BITS), %% assert\n-  CorrId = <<ApiKey:?API_KEY_BITS,\n-             Vsn:?API_VERSION_BITS,\n-             CorrId0:?CORR_ID_BITS>>,\n-  IoData =\n-    [ encode(int16, ApiKey)\n-    , encode(int16, Vsn)\n-    , CorrId\n-    , encode(string, ClientId)\n-    , encode_struct(Tag, Vsn, Msg)\n-    ],\n-  Size = data_size(IoData),\n-  [encode(int32, Size), IoData].\n-\n-%% @doc Parse binary stream received from kafka broker.\n-%% Return a list of kpro:rsp() and the remaining bytes.\n-%% @end\n--spec decode_response(binary()) -> {[rsp()], binary()}.\n-decode_response(Bin) ->\n-  decode_response(Bin, []).\n+encode_request(ClientId, CorrId, Req) ->\n+  kpro_req_lib:encode(ClientId, CorrId, Req).\n \n %% @doc The messageset is not decoded upon receiving (in socket process).\n %% Pass the message set as binary to the consumer process and decode there\n-%% Return {incomplete_message, ExpectedSize} if the fetch size is not big\n-%% enough for even one single message. Otherwise return a list of decoded\n-%% messages.\n-%% @end\n--spec decode_message_set(binary()) -> incomplete_message() | [message()].\n-decode_message_set(MessageSetBin) when is_binary(MessageSetBin) ->\n-  case decode_message_stream(MessageSetBin, []) of\n-    [?incomplete_message(_) = Incomplete] ->\n-      %% The only message is incomplete\n-      %% return the special tuple\n-      Incomplete;\n-    [?incomplete_message(_) | Messages] ->\n-      %% Discard the last incomplete message\n-      lists:reverse(Messages);\n-    Messages ->\n-      lists:reverse(Messages)\n-  end.\n-\n-%%%_* Hidden APIs ==============================================================\n-\n-%% @hidden Encode prmitives.\n--spec encode(primitive_type(), primitive()) -> iodata().\n-encode(boolean, true) -> <<1:8/?INT>>;\n-encode(boolean, false) -> <<0:8/?INT>>;\n-encode(int8,  I) when is_integer(I) -> <<I:8/?INT>>;\n-encode(int16, I) when is_integer(I) -> <<I:16/?INT>>;\n-encode(int32, I) when is_integer(I) -> <<I:32/?INT>>;\n-encode(int64, I) when is_integer(I) -> <<I:64/?INT>>;\n-encode(nullable_string, ?null) -> <<-1:16/?INT>>;\n-encode(nullable_string, Str) -> encode(string, Str);\n-encode(string, Atom) when is_atom(Atom) ->\n-  encode(string, atom_to_binary(Atom, utf8));\n-encode(string, <<>>) -> <<0:16/?INT>>;\n-encode(string, L) when is_list(L) ->\n-  encode(string, bin(L));\n-encode(string, B) when is_binary(B) ->\n-  Length = size(B),\n-  <<Length:16/?INT, B/binary>>;\n-encode(bytes, ?null) -> <<-1:32/?INT>>;\n-encode(bytes, B) when is_binary(B) orelse is_list(B) ->\n-  Size = data_size(B),\n-  case Size =:= 0 of\n-    true  -> <<-1:32/?INT>>;\n-    false -> [<<Size:32/?INT>>, B]\n-  end;\n-encode(records, B) ->\n-  encode(bytes, B).\n-\n-%% @hidden Decode prmitives.\n--spec decode(primitive_type(), binary()) -> {primitive(), binary()}.\n-decode(boolean, Bin) ->\n-  <<Value:8/?INT, Rest/binary>> = Bin,\n-  {Value =/= 0, Rest};\n-decode(int8, Bin) ->\n-  <<Value:8/?INT, Rest/binary>> = Bin,\n-  {Value, Rest};\n-decode(int16, Bin) ->\n-  <<Value:16/?INT, Rest/binary>> = Bin,\n-  {Value, Rest};\n-decode(int32, Bin) ->\n-  <<Value:32/?INT, Rest/binary>> = Bin,\n-  {Value, Rest};\n-decode(int64, Bin) ->\n-  <<Value:64/?INT, Rest/binary>> = Bin,\n-  {Value, Rest};\n-decode(string, Bin) ->\n-  <<Size:16/?INT, Rest/binary>> = Bin,\n-  copy_bytes(Size, Rest);\n-decode(bytes, Bin) ->\n-  <<Size:32/?INT, Rest/binary>> = Bin,\n-  copy_bytes(Size, Rest);\n-decode(nullable_string, Bin) ->\n-  decode(string, Bin);\n-decode(records, Bin) ->\n-  decode(bytes, Bin).\n-\n-%% @hidden Encode struct.\n--spec enc_struct(schema(), struct(), stack()) -> iodata().\n-enc_struct([], _Values, _Stack) -> [];\n-enc_struct([{Name, FieldSc} | Schema], Values, Stack) when is_list(Values) ->\n-  NewStack = [Name | Stack],\n-  case lists:keytake(Name, 1, Values) of\n-    {value, {_, Value0}, ValuesLeft} ->\n-      Value = enc_embedded(NewStack, Value0),\n-      [ enc_struct_field(FieldSc, Value, NewStack)\n-      | enc_struct(Schema, ValuesLeft, Stack)\n-      ];\n-    false ->\n-      erlang:throw({field_missing, [Name | Stack]})\n-  end;\n-enc_struct(_Schema, _Value, Stack) ->\n-  erlang:throw({not_struct, Stack}).\n-\n-%% @hidden Decode struct.\n--spec dec_struct(struct_schema(), struct(), stack(), binary()) ->\n-        {struct(), binary()}.\n-dec_struct([], Fields, _Stack, Bin) ->\n-  {lists:reverse(Fields), Bin};\n-dec_struct([{Name, FieldSc} | Schema], Fields, Stack, Bin) ->\n-  NewStack = [Name | Stack],\n-  {Value0, Rest} = dec_struct_field(FieldSc, NewStack, Bin),\n-  Value = dec_embedded(NewStack, Value0),\n-  dec_struct(Schema, [{Name, Value} | Fields], Stack, Rest).\n-\n-%% @hidden Encode struct having schema predefined in kpro_schema.\n--spec encode_struct(req_tag(), vsn(), binary() | struct()) -> binary().\n-encode_struct(Tag, Vsn, Bin) ->\n-  encode_struct(?SCHEMA_MODULE, Tag, Vsn, Bin).\n-\n-%% @hidden Encode struct having schema predefined in a callback:\n-%% Module:get(Tag, Vsn)\n-%% @end\n--spec encode_struct(module(), req_tag(), vsn(),\n-                    binary() | struct()) -> binary().\n-encode_struct(_Module, _Tag, _Vsn, Bin) when is_binary(Bin) -> Bin;\n-encode_struct(Module, Tag, Vsn, Fields) ->\n-  Schema = get_schema(Module, Tag, Vsn),\n-  try\n-    bin(enc_struct(Schema, Fields, [{Tag, Vsn}]))\n-  catch\n-    throw : {Reason, Stack} ->\n-      Trace = erlang:get_stacktrace(),\n-      erlang:raise(error, {Reason, Stack, Fields}, Trace)\n-  end.\n-\n-%% @hidden Decode struct having schema predefined in kpro_schema.\n--spec decode_struct(rsp_tag(), vsn(), binary()) ->\n-        {struct(), binary()}.\n-decode_struct(Tag, Vsn, Bin) ->\n-  decode_struct(?SCHEMA_MODULE, Tag, Vsn, Bin).\n-\n-%% @hidden Decode struct having schema predefined in a callback:\n-%% Module:get(Tag, Vsn)\n-%% @end\n--spec decode_struct(module(), rsp_tag(), vsn(), binary()) ->\n-        {struct(), binary()}.\n-decode_struct(Module, Tag, Vsn, Bin) ->\n-  Schema = get_schema(Module, Tag, Vsn),\n-  dec_struct(Schema, _Fields = [], _Stack = [{Tag, Vsn}], Bin).\n-\n-%% @hidden Get predefined schema from kpro_schema:get/2.\n--spec get_schema(tag(), vsn()) -> struct_schema().\n-get_schema(Tag, Vsn) ->\n-  get_schema(?SCHEMA_MODULE, Tag, Vsn).\n-\n-%% @hidden Get predefined schema from Module:get/2 API.\n--spec get_schema(module(), tag(), vsn()) -> struct_schema().\n-get_schema(Module, Tag, Vsn) ->\n-  try\n-    Module:get(Tag, Vsn)\n-  catch\n-    error : function_clause when Vsn =:= 0 ->\n-      erlang:error({unknown_tag, Tag});\n-    error : function_clause when Vsn > 0 ->\n-      try\n-        _ = Module:get(Tag, 0)\n-      catch\n-        error : function_clause ->\n-          erlang:error({unknown_tag, Tag})\n-      end,\n-      erlang:error({unsupported_version, Tag, Vsn})\n-  end.\n-\n-%% @doc Find field value in a struct, raise an exception if not found.\n--spec find(field_name(), struct()) -> field_value() | no_return().\n+%% Return `?incomplete_batch(ExpectedSize)' if the fetch size is not big\n+%% enough for even one single message. Otherwise return `{Meta, Messages}'\n+%% where `Meta' is either `?KPRO_NO_BATCH_META' for magic-version 0-1 or\n+%% `#kafka_batch_meta{}' for magic-version 2 or above.\n+-spec decode_batches(binary()) -> kpro:batch_decode_result().\n+decode_batches(<<_:64/?INT, L:32, T/binary>> = Bin) when size(T) >= L ->\n+  kpro_batch:decode(Bin);\n+decode_batches(<<_:64/?INT, L:32, _T/binary>>) ->\n+  %% not enough to decode one single message for magic v0-1\n+  ?incomplete_batch(L + ?BATCH_LEADING_BYTES);\n+decode_batches(_) ->\n+  %% not enough to even get the size header\n+  ?incomplete_batch(?BATCH_LEADING_BYTES).\n+\n+%% @doc Send a request, wait for response.\n+%% Immediately return 'ok' if it is a produce request with `required_acks = 0'.\n+-spec request_sync(pid(), req(), timeout()) ->\n+        ok | {ok, rsp()} | {error, any()}.\n+request_sync(ConnectionPid, Request, Timeout) ->\n+  kpro_connection:request_sync(ConnectionPid, Request, Timeout).\n+\n+%% @doc Send a request and get back a correlation ID to match future response.\n+%% Immediately return 'ok' if it is a produce request with `required_acks = 0'.\n+-spec request_async(pid(), req()) ->\n+        ok | {ok, corr_id()} | {error, any()}.\n+request_async(ConnectionPid, Request) ->\n+  kpro_connection:request_async(ConnectionPid, Request).\n+\n+%% @doc Connect to any of the endpoints in the given list.\n+%% NOTE: Connection process is linked to caller unless `nolink => true'\n+%%       is set in connection config\n+-spec connect_any([endpoint()], conn_config()) ->\n+        {ok, connection()} | {error, any()}.\n+connect_any(Endpoints, ConnConfig) ->\n+  kpro_brokers:connect_any(Endpoints, ConnConfig).\n+\n+%% @doc Sotp connection process.\n+-spec close_connection(connection()) -> ok.\n+close_connection(Connection) ->\n+  kpro_connection:stop(Connection).\n+\n+%% @doc Connect partition leader.\n+%% If the fist arg is not an already established metadata connection\n+%% but a bootstraping endpoint list, this function will first try to\n+%% establish a temp connection to any of the bootstraping endpoints.","path":"src/kpro.erl","position":669,"original_position":668,"commit_id":"b0862032a0376c87084c70a12cb71e1afb9c54ea","original_commit_id":"dbbd849f5905cd6eccd2bcbb95bbe676bb4ac588","user":{"login":"mikpe","id":7581288,"avatar_url":"https://avatars3.githubusercontent.com/u/7581288?v=4","gravatar_id":"","url":"https://api.github.com/users/mikpe","html_url":"https://github.com/mikpe","followers_url":"https://api.github.com/users/mikpe/followers","following_url":"https://api.github.com/users/mikpe/following{/other_user}","gists_url":"https://api.github.com/users/mikpe/gists{/gist_id}","starred_url":"https://api.github.com/users/mikpe/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mikpe/subscriptions","organizations_url":"https://api.github.com/users/mikpe/orgs","repos_url":"https://api.github.com/users/mikpe/repos","events_url":"https://api.github.com/users/mikpe/events{/privacy}","received_events_url":"https://api.github.com/users/mikpe/received_events","type":"User","site_admin":false},"body":"Nit: bootstraping","created_at":"2018-05-02T08:41:25Z","updated_at":"2018-05-02T08:41:25Z","html_url":"https://github.com/klarna/kafka_protocol/pull/40#discussion_r185427815","pull_request_url":"https://api.github.com/repos/klarna/kafka_protocol/pulls/40","author_association":"NONE","_links":{"self":{"href":"https://api.github.com/repos/klarna/kafka_protocol/pulls/comments/185427815"},"html":{"href":"https://github.com/klarna/kafka_protocol/pull/40#discussion_r185427815"},"pull_request":{"href":"https://api.github.com/repos/klarna/kafka_protocol/pulls/40"}}},"pull_request":{"url":"https://api.github.com/repos/klarna/kafka_protocol/pulls/40","id":184963990,"html_url":"https://github.com/klarna/kafka_protocol/pull/40","diff_url":"https://github.com/klarna/kafka_protocol/pull/40.diff","patch_url":"https://github.com/klarna/kafka_protocol/pull/40.patch","issue_url":"https://api.github.com/repos/klarna/kafka_protocol/issues/40","number":40,"state":"open","locked":false,"title":"Support kafka protocol 1.1","user":{"login":"zmstone","id":164324,"avatar_url":"https://avatars3.githubusercontent.com/u/164324?v=4","gravatar_id":"","url":"https://api.github.com/users/zmstone","html_url":"https://github.com/zmstone","followers_url":"https://api.github.com/users/zmstone/followers","following_url":"https://api.github.com/users/zmstone/following{/other_user}","gists_url":"https://api.github.com/users/zmstone/gists{/gist_id}","starred_url":"https://api.github.com/users/zmstone/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zmstone/subscriptions","organizations_url":"https://api.github.com/users/zmstone/orgs","repos_url":"https://api.github.com/users/zmstone/repos","events_url":"https://api.github.com/users/zmstone/events{/privacy}","received_events_url":"https://api.github.com/users/zmstone/received_events","type":"User","site_admin":false},"body":"Aiming for a `2.0.0` release. i.e. non-compatible APIs vs. `1.2.0`\r\nMain changes:\r\n* Supported kafka 1.1 protocol\r\n* API keys are generated from bnf\r\n* Error codes are generated from a eterm.\r\n* Schema getters changed from `get(xxx_request, Vsn)` to `req(xxx, Vsn)`, and `get(yyy_response, Vsn)` to `rsp(yyy, Vsn)`. \r\n* `kpro:struct()` is now a `map()`, but `list()` is still supported as encoder input\r\n* Added socket implementation `kpro_connection.erl` (a copy of `brod_sock`).\r\n* Basic connection management APIs (connect partition-leader, group-coordinator, transactional-coordinator, cluster-controller)\r\n* Transactional RPC primitives `kpro:txn_xxx`\r\n* Changed socket option from `{packet, raw}` to `{packet, 4}`\r\n","created_at":"2018-04-30T14:00:32Z","updated_at":"2018-05-02T08:41:25Z","closed_at":null,"merged_at":null,"merge_commit_sha":"3ef5eecaa2de3d5294dbd28fc4b777dc123d28af","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/klarna/kafka_protocol/pulls/40/commits","review_comments_url":"https://api.github.com/repos/klarna/kafka_protocol/pulls/40/comments","review_comment_url":"https://api.github.com/repos/klarna/kafka_protocol/pulls/comments{/number}","comments_url":"https://api.github.com/repos/klarna/kafka_protocol/issues/40/comments","statuses_url":"https://api.github.com/repos/klarna/kafka_protocol/statuses/b0862032a0376c87084c70a12cb71e1afb9c54ea","head":{"label":"klarna:support-kafka-1.0.0","ref":"support-kafka-1.0.0","sha":"b0862032a0376c87084c70a12cb71e1afb9c54ea","user":{"login":"klarna","id":394540,"avatar_url":"https://avatars0.githubusercontent.com/u/394540?v=4","gravatar_id":"","url":"https://api.github.com/users/klarna","html_url":"https://github.com/klarna","followers_url":"https://api.github.com/users/klarna/followers","following_url":"https://api.github.com/users/klarna/following{/other_user}","gists_url":"https://api.github.com/users/klarna/gists{/gist_id}","starred_url":"https://api.github.com/users/klarna/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/klarna/subscriptions","organizations_url":"https://api.github.com/users/klarna/orgs","repos_url":"https://api.github.com/users/klarna/repos","events_url":"https://api.github.com/users/klarna/events{/privacy}","received_events_url":"https://api.github.com/users/klarna/received_events","type":"Organization","site_admin":false},"repo":{"id":52611237,"name":"kafka_protocol","full_name":"klarna/kafka_protocol","owner":{"login":"klarna","id":394540,"avatar_url":"https://avatars0.githubusercontent.com/u/394540?v=4","gravatar_id":"","url":"https://api.github.com/users/klarna","html_url":"https://github.com/klarna","followers_url":"https://api.github.com/users/klarna/followers","following_url":"https://api.github.com/users/klarna/following{/other_user}","gists_url":"https://api.github.com/users/klarna/gists{/gist_id}","starred_url":"https://api.github.com/users/klarna/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/klarna/subscriptions","organizations_url":"https://api.github.com/users/klarna/orgs","repos_url":"https://api.github.com/users/klarna/repos","events_url":"https://api.github.com/users/klarna/events{/privacy}","received_events_url":"https://api.github.com/users/klarna/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/klarna/kafka_protocol","description":"Kafka protocol erlang library","fork":false,"url":"https://api.github.com/repos/klarna/kafka_protocol","forks_url":"https://api.github.com/repos/klarna/kafka_protocol/forks","keys_url":"https://api.github.com/repos/klarna/kafka_protocol/keys{/key_id}","collaborators_url":"https://api.github.com/repos/klarna/kafka_protocol/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/klarna/kafka_protocol/teams","hooks_url":"https://api.github.com/repos/klarna/kafka_protocol/hooks","issue_events_url":"https://api.github.com/repos/klarna/kafka_protocol/issues/events{/number}","events_url":"https://api.github.com/repos/klarna/kafka_protocol/events","assignees_url":"https://api.github.com/repos/klarna/kafka_protocol/assignees{/user}","branches_url":"https://api.github.com/repos/klarna/kafka_protocol/branches{/branch}","tags_url":"https://api.github.com/repos/klarna/kafka_protocol/tags","blobs_url":"https://api.github.com/repos/klarna/kafka_protocol/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/klarna/kafka_protocol/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/klarna/kafka_protocol/git/refs{/sha}","trees_url":"https://api.github.com/repos/klarna/kafka_protocol/git/trees{/sha}","statuses_url":"https://api.github.com/repos/klarna/kafka_protocol/statuses/{sha}","languages_url":"https://api.github.com/repos/klarna/kafka_protocol/languages","stargazers_url":"https://api.github.com/repos/klarna/kafka_protocol/stargazers","contributors_url":"https://api.github.com/repos/klarna/kafka_protocol/contributors","subscribers_url":"https://api.github.com/repos/klarna/kafka_protocol/subscribers","subscription_url":"https://api.github.com/repos/klarna/kafka_protocol/subscription","commits_url":"https://api.github.com/repos/klarna/kafka_protocol/commits{/sha}","git_commits_url":"https://api.github.com/repos/klarna/kafka_protocol/git/commits{/sha}","comments_url":"https://api.github.com/repos/klarna/kafka_protocol/comments{/number}","issue_comment_url":"https://api.github.com/repos/klarna/kafka_protocol/issues/comments{/number}","contents_url":"https://api.github.com/repos/klarna/kafka_protocol/contents/{+path}","compare_url":"https://api.github.com/repos/klarna/kafka_protocol/compare/{base}...{head}","merges_url":"https://api.github.com/repos/klarna/kafka_protocol/merges","archive_url":"https://api.github.com/repos/klarna/kafka_protocol/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/klarna/kafka_protocol/downloads","issues_url":"https://api.github.com/repos/klarna/kafka_protocol/issues{/number}","pulls_url":"https://api.github.com/repos/klarna/kafka_protocol/pulls{/number}","milestones_url":"https://api.github.com/repos/klarna/kafka_protocol/milestones{/number}","notifications_url":"https://api.github.com/repos/klarna/kafka_protocol/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/klarna/kafka_protocol/labels{/name}","releases_url":"https://api.github.com/repos/klarna/kafka_protocol/releases{/id}","deployments_url":"https://api.github.com/repos/klarna/kafka_protocol/deployments","created_at":"2016-02-26T14:52:38Z","updated_at":"2018-04-03T20:47:56Z","pushed_at":"2018-05-02T08:25:01Z","git_url":"git://github.com/klarna/kafka_protocol.git","ssh_url":"git@github.com:klarna/kafka_protocol.git","clone_url":"https://github.com/klarna/kafka_protocol.git","svn_url":"https://github.com/klarna/kafka_protocol","homepage":null,"size":459,"stargazers_count":21,"watchers_count":21,"language":"Erlang","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":10,"mirror_url":null,"archived":false,"open_issues_count":1,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":10,"open_issues":1,"watchers":21,"default_branch":"master"}},"base":{"label":"klarna:master","ref":"master","sha":"6af2802a96ce1112f0fb613b775b8f7ffc3655d8","user":{"login":"klarna","id":394540,"avatar_url":"https://avatars0.githubusercontent.com/u/394540?v=4","gravatar_id":"","url":"https://api.github.com/users/klarna","html_url":"https://github.com/klarna","followers_url":"https://api.github.com/users/klarna/followers","following_url":"https://api.github.com/users/klarna/following{/other_user}","gists_url":"https://api.github.com/users/klarna/gists{/gist_id}","starred_url":"https://api.github.com/users/klarna/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/klarna/subscriptions","organizations_url":"https://api.github.com/users/klarna/orgs","repos_url":"https://api.github.com/users/klarna/repos","events_url":"https://api.github.com/users/klarna/events{/privacy}","received_events_url":"https://api.github.com/users/klarna/received_events","type":"Organization","site_admin":false},"repo":{"id":52611237,"name":"kafka_protocol","full_name":"klarna/kafka_protocol","owner":{"login":"klarna","id":394540,"avatar_url":"https://avatars0.githubusercontent.com/u/394540?v=4","gravatar_id":"","url":"https://api.github.com/users/klarna","html_url":"https://github.com/klarna","followers_url":"https://api.github.com/users/klarna/followers","following_url":"https://api.github.com/users/klarna/following{/other_user}","gists_url":"https://api.github.com/users/klarna/gists{/gist_id}","starred_url":"https://api.github.com/users/klarna/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/klarna/subscriptions","organizations_url":"https://api.github.com/users/klarna/orgs","repos_url":"https://api.github.com/users/klarna/repos","events_url":"https://api.github.com/users/klarna/events{/privacy}","received_events_url":"https://api.github.com/users/klarna/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/klarna/kafka_protocol","description":"Kafka protocol erlang library","fork":false,"url":"https://api.github.com/repos/klarna/kafka_protocol","forks_url":"https://api.github.com/repos/klarna/kafka_protocol/forks","keys_url":"https://api.github.com/repos/klarna/kafka_protocol/keys{/key_id}","collaborators_url":"https://api.github.com/repos/klarna/kafka_protocol/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/klarna/kafka_protocol/teams","hooks_url":"https://api.github.com/repos/klarna/kafka_protocol/hooks","issue_events_url":"https://api.github.com/repos/klarna/kafka_protocol/issues/events{/number}","events_url":"https://api.github.com/repos/klarna/kafka_protocol/events","assignees_url":"https://api.github.com/repos/klarna/kafka_protocol/assignees{/user}","branches_url":"https://api.github.com/repos/klarna/kafka_protocol/branches{/branch}","tags_url":"https://api.github.com/repos/klarna/kafka_protocol/tags","blobs_url":"https://api.github.com/repos/klarna/kafka_protocol/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/klarna/kafka_protocol/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/klarna/kafka_protocol/git/refs{/sha}","trees_url":"https://api.github.com/repos/klarna/kafka_protocol/git/trees{/sha}","statuses_url":"https://api.github.com/repos/klarna/kafka_protocol/statuses/{sha}","languages_url":"https://api.github.com/repos/klarna/kafka_protocol/languages","stargazers_url":"https://api.github.com/repos/klarna/kafka_protocol/stargazers","contributors_url":"https://api.github.com/repos/klarna/kafka_protocol/contributors","subscribers_url":"https://api.github.com/repos/klarna/kafka_protocol/subscribers","subscription_url":"https://api.github.com/repos/klarna/kafka_protocol/subscription","commits_url":"https://api.github.com/repos/klarna/kafka_protocol/commits{/sha}","git_commits_url":"https://api.github.com/repos/klarna/kafka_protocol/git/commits{/sha}","comments_url":"https://api.github.com/repos/klarna/kafka_protocol/comments{/number}","issue_comment_url":"https://api.github.com/repos/klarna/kafka_protocol/issues/comments{/number}","contents_url":"https://api.github.com/repos/klarna/kafka_protocol/contents/{+path}","compare_url":"https://api.github.com/repos/klarna/kafka_protocol/compare/{base}...{head}","merges_url":"https://api.github.com/repos/klarna/kafka_protocol/merges","archive_url":"https://api.github.com/repos/klarna/kafka_protocol/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/klarna/kafka_protocol/downloads","issues_url":"https://api.github.com/repos/klarna/kafka_protocol/issues{/number}","pulls_url":"https://api.github.com/repos/klarna/kafka_protocol/pulls{/number}","milestones_url":"https://api.github.com/repos/klarna/kafka_protocol/milestones{/number}","notifications_url":"https://api.github.com/repos/klarna/kafka_protocol/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/klarna/kafka_protocol/labels{/name}","releases_url":"https://api.github.com/repos/klarna/kafka_protocol/releases{/id}","deployments_url":"https://api.github.com/repos/klarna/kafka_protocol/deployments","created_at":"2016-02-26T14:52:38Z","updated_at":"2018-04-03T20:47:56Z","pushed_at":"2018-05-02T08:25:01Z","git_url":"git://github.com/klarna/kafka_protocol.git","ssh_url":"git@github.com:klarna/kafka_protocol.git","clone_url":"https://github.com/klarna/kafka_protocol.git","svn_url":"https://github.com/klarna/kafka_protocol","homepage":null,"size":459,"stargazers_count":21,"watchers_count":21,"language":"Erlang","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":10,"mirror_url":null,"archived":false,"open_issues_count":1,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":10,"open_issues":1,"watchers":21,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/klarna/kafka_protocol/pulls/40"},"html":{"href":"https://github.com/klarna/kafka_protocol/pull/40"},"issue":{"href":"https://api.github.com/repos/klarna/kafka_protocol/issues/40"},"comments":{"href":"https://api.github.com/repos/klarna/kafka_protocol/issues/40/comments"},"review_comments":{"href":"https://api.github.com/repos/klarna/kafka_protocol/pulls/40/comments"},"review_comment":{"href":"https://api.github.com/repos/klarna/kafka_protocol/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/klarna/kafka_protocol/pulls/40/commits"},"statuses":{"href":"https://api.github.com/repos/klarna/kafka_protocol/statuses/b0862032a0376c87084c70a12cb71e1afb9c54ea"}},"author_association":"MEMBER"}}
{"actor":{"display_login":"mikpe"}}