Packages

An Elixir implementation of the Kafka protocol. It enables communication with Kafka brokers using standard Elixir data structures without the need for manual serialization.

Current section

Files

Jump to
klife_protocol lib klife_protocol generated messages join_group.ex
Raw

lib/klife_protocol/generated/messages/join_group.ex

# DO NOT EDIT THIS FILE MANUALLY
# This module is automatically generated by running mix task generate_file
# every change must be done inside the mix task directly
defmodule KlifeProtocol.Messages.JoinGroup do
@moduledoc """
Kafka protocol JoinGroup message
Request versions summary:
- Version 1 adds RebalanceTimeoutMs. Version 2 and 3 are the same as version 1.
- Starting from version 4, the client needs to issue a second request to join group
- Starting from version 5, we add a new field called groupInstanceId to indicate member identity across restarts.
with assigned id.
- Version 6 is the first flexible version.
- Version 7 is the same as version 6.
- Version 8 adds the Reason field (KIP-800).
- Version 9 is the same as version 8.
Response versions summary:
- Version 1 is the same as version 0.
- Version 2 adds throttle time.
- Starting in version 3, on quota violation, brokers send out responses before throttling.
- Starting in version 4, the client needs to issue a second request to join group
with assigned id.
- Version 5 is bumped to apply group.instance.id to identify member across restarts.
- Version 6 is the first flexible version.
- Starting from version 7, the broker sends back the Protocol Type to the client (KIP-559).
- Version 8 is the same as version 7.
- Version 9 adds the SkipAssignment field.
"""
alias KlifeProtocol.Deserializer
alias KlifeProtocol.Serializer
alias KlifeProtocol.Header
@api_key 11
@min_flexible_version_req 6
@min_flexible_version_res 6
@doc """
Receives a map and serialize it to kafka wire format of the given version.
Input content fields:
- group_id: The group identifier. (string | versions 0+)
- session_timeout_ms: The coordinator considers the consumer dead if it receives no heartbeat after this timeout in milliseconds. (int32 | versions 0+)
- rebalance_timeout_ms: The maximum time in milliseconds that the coordinator will wait for each member to rejoin when rebalancing the group. (int32 | versions 1+)
- member_id: The member id assigned by the group coordinator. (string | versions 0+)
- group_instance_id: The unique identifier of the consumer instance provided by end user. (string | versions 5+)
- protocol_type: The unique name the for class of protocols implemented by the group we want to join. (string | versions 0+)
- protocols: The list of protocols that the member supports. ([]JoinGroupRequestProtocol | versions 0+)
- name: The protocol name. (string | versions 0+)
- metadata: The protocol metadata. (bytes | versions 0+)
- reason: The reason why the member (re-)joins the group. (string | versions 8+)
"""
def serialize_request(%{headers: headers, content: content}, version) do
headers
|> Map.put(:request_api_key, @api_key)
|> Map.put(:request_api_version, version)
|> Header.serialize_request(req_header_version(version))
|> then(&Serializer.execute(content, request_schema(version), &1))
end
@doc """
Receive a binary in the kafka wire format and deserialize it into a map.
Response content fields:
- throttle_time_ms: The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota. (int32 | versions 2+)
- error_code: The error code, or 0 if there was no error. (int16 | versions 0+)
- generation_id: The generation ID of the group. (int32 | versions 0+)
- protocol_type: The group protocol name. (string | versions 7+)
- protocol_name: The group protocol selected by the coordinator. (string | versions 0+)
- leader: The leader of the group. (string | versions 0+)
- skip_assignment: True if the leader must skip running the assignment. (bool | versions 9+)
- member_id: The member ID assigned by the group coordinator. (string | versions 0+)
- members: The group members. ([]JoinGroupResponseMember | versions 0+)
- member_id: The group member ID. (string | versions 0+)
- group_instance_id: The unique identifier of the consumer instance provided by end user. (string | versions 5+)
- metadata: The group member metadata. (bytes | versions 0+)
"""
def deserialize_response(data, version, with_header? \\ true)
def deserialize_response(data, version, true) do
{:ok, {headers, rest_data}} = Header.deserialize_response(data, res_header_version(version))
case Deserializer.execute(rest_data, response_schema(version)) do
{:ok, {content, <<>>}} ->
{:ok, %{headers: headers, content: content}}
{:error, _reason} = err ->
err
end
end
def deserialize_response(data, version, false) do
case Deserializer.execute(data, response_schema(version)) do
{:ok, {content, <<>>}} ->
{:ok, %{content: content}}
{:error, _reason} = err ->
err
end
end
@doc """
Returns the message api key number.
"""
def api_key(), do: @api_key
@doc """
Returns the current max supported version of this message.
"""
def max_supported_version(), do: 9
@doc """
Returns the current min supported version of this message.
"""
def min_supported_version(), do: 0
defp req_header_version(msg_version),
do: if(msg_version >= @min_flexible_version_req, do: 2, else: 1)
defp res_header_version(msg_version),
do: if(msg_version >= @min_flexible_version_res, do: 1, else: 0)
def request_schema(0),
do: [
group_id: {:string, %{is_nullable?: false}},
session_timeout_ms: {:int32, %{is_nullable?: false}},
member_id: {:string, %{is_nullable?: false}},
protocol_type: {:string, %{is_nullable?: false}},
protocols:
{{:array,
[name: {:string, %{is_nullable?: false}}, metadata: {:bytes, %{is_nullable?: false}}]},
%{is_nullable?: false}}
]
def request_schema(1),
do: [
group_id: {:string, %{is_nullable?: false}},
session_timeout_ms: {:int32, %{is_nullable?: false}},
rebalance_timeout_ms: {:int32, %{is_nullable?: false}},
member_id: {:string, %{is_nullable?: false}},
protocol_type: {:string, %{is_nullable?: false}},
protocols:
{{:array,
[name: {:string, %{is_nullable?: false}}, metadata: {:bytes, %{is_nullable?: false}}]},
%{is_nullable?: false}}
]
def request_schema(2),
do: [
group_id: {:string, %{is_nullable?: false}},
session_timeout_ms: {:int32, %{is_nullable?: false}},
rebalance_timeout_ms: {:int32, %{is_nullable?: false}},
member_id: {:string, %{is_nullable?: false}},
protocol_type: {:string, %{is_nullable?: false}},
protocols:
{{:array,
[name: {:string, %{is_nullable?: false}}, metadata: {:bytes, %{is_nullable?: false}}]},
%{is_nullable?: false}}
]
def request_schema(3),
do: [
group_id: {:string, %{is_nullable?: false}},
session_timeout_ms: {:int32, %{is_nullable?: false}},
rebalance_timeout_ms: {:int32, %{is_nullable?: false}},
member_id: {:string, %{is_nullable?: false}},
protocol_type: {:string, %{is_nullable?: false}},
protocols:
{{:array,
[name: {:string, %{is_nullable?: false}}, metadata: {:bytes, %{is_nullable?: false}}]},
%{is_nullable?: false}}
]
def request_schema(4),
do: [
group_id: {:string, %{is_nullable?: false}},
session_timeout_ms: {:int32, %{is_nullable?: false}},
rebalance_timeout_ms: {:int32, %{is_nullable?: false}},
member_id: {:string, %{is_nullable?: false}},
protocol_type: {:string, %{is_nullable?: false}},
protocols:
{{:array,
[name: {:string, %{is_nullable?: false}}, metadata: {:bytes, %{is_nullable?: false}}]},
%{is_nullable?: false}}
]
def request_schema(5),
do: [
group_id: {:string, %{is_nullable?: false}},
session_timeout_ms: {:int32, %{is_nullable?: false}},
rebalance_timeout_ms: {:int32, %{is_nullable?: false}},
member_id: {:string, %{is_nullable?: false}},
group_instance_id: {:string, %{is_nullable?: true}},
protocol_type: {:string, %{is_nullable?: false}},
protocols:
{{:array,
[name: {:string, %{is_nullable?: false}}, metadata: {:bytes, %{is_nullable?: false}}]},
%{is_nullable?: false}}
]
def request_schema(6),
do: [
group_id: {:compact_string, %{is_nullable?: false}},
session_timeout_ms: {:int32, %{is_nullable?: false}},
rebalance_timeout_ms: {:int32, %{is_nullable?: false}},
member_id: {:compact_string, %{is_nullable?: false}},
group_instance_id: {:compact_string, %{is_nullable?: true}},
protocol_type: {:compact_string, %{is_nullable?: false}},
protocols:
{{:compact_array,
[
name: {:compact_string, %{is_nullable?: false}},
metadata: {:compact_bytes, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, []}
]}, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, []}
]
def request_schema(7),
do: [
group_id: {:compact_string, %{is_nullable?: false}},
session_timeout_ms: {:int32, %{is_nullable?: false}},
rebalance_timeout_ms: {:int32, %{is_nullable?: false}},
member_id: {:compact_string, %{is_nullable?: false}},
group_instance_id: {:compact_string, %{is_nullable?: true}},
protocol_type: {:compact_string, %{is_nullable?: false}},
protocols:
{{:compact_array,
[
name: {:compact_string, %{is_nullable?: false}},
metadata: {:compact_bytes, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, []}
]}, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, []}
]
def request_schema(8),
do: [
group_id: {:compact_string, %{is_nullable?: false}},
session_timeout_ms: {:int32, %{is_nullable?: false}},
rebalance_timeout_ms: {:int32, %{is_nullable?: false}},
member_id: {:compact_string, %{is_nullable?: false}},
group_instance_id: {:compact_string, %{is_nullable?: true}},
protocol_type: {:compact_string, %{is_nullable?: false}},
protocols:
{{:compact_array,
[
name: {:compact_string, %{is_nullable?: false}},
metadata: {:compact_bytes, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, []}
]}, %{is_nullable?: false}},
reason: {:compact_string, %{is_nullable?: true}},
tag_buffer: {:tag_buffer, []}
]
def request_schema(9),
do: [
group_id: {:compact_string, %{is_nullable?: false}},
session_timeout_ms: {:int32, %{is_nullable?: false}},
rebalance_timeout_ms: {:int32, %{is_nullable?: false}},
member_id: {:compact_string, %{is_nullable?: false}},
group_instance_id: {:compact_string, %{is_nullable?: true}},
protocol_type: {:compact_string, %{is_nullable?: false}},
protocols:
{{:compact_array,
[
name: {:compact_string, %{is_nullable?: false}},
metadata: {:compact_bytes, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, []}
]}, %{is_nullable?: false}},
reason: {:compact_string, %{is_nullable?: true}},
tag_buffer: {:tag_buffer, []}
]
def request_schema(unkown_version),
do: raise("Unknown version #{unkown_version} for message JoinGroup")
def response_schema(0),
do: [
error_code: {:int16, %{is_nullable?: false}},
generation_id: {:int32, %{is_nullable?: false}},
protocol_name: {:string, %{is_nullable?: false}},
leader: {:string, %{is_nullable?: false}},
member_id: {:string, %{is_nullable?: false}},
members:
{{:array,
[
member_id: {:string, %{is_nullable?: false}},
metadata: {:bytes, %{is_nullable?: false}}
]}, %{is_nullable?: false}}
]
def response_schema(1),
do: [
error_code: {:int16, %{is_nullable?: false}},
generation_id: {:int32, %{is_nullable?: false}},
protocol_name: {:string, %{is_nullable?: false}},
leader: {:string, %{is_nullable?: false}},
member_id: {:string, %{is_nullable?: false}},
members:
{{:array,
[
member_id: {:string, %{is_nullable?: false}},
metadata: {:bytes, %{is_nullable?: false}}
]}, %{is_nullable?: false}}
]
def response_schema(2),
do: [
throttle_time_ms: {:int32, %{is_nullable?: false}},
error_code: {:int16, %{is_nullable?: false}},
generation_id: {:int32, %{is_nullable?: false}},
protocol_name: {:string, %{is_nullable?: false}},
leader: {:string, %{is_nullable?: false}},
member_id: {:string, %{is_nullable?: false}},
members:
{{:array,
[
member_id: {:string, %{is_nullable?: false}},
metadata: {:bytes, %{is_nullable?: false}}
]}, %{is_nullable?: false}}
]
def response_schema(3),
do: [
throttle_time_ms: {:int32, %{is_nullable?: false}},
error_code: {:int16, %{is_nullable?: false}},
generation_id: {:int32, %{is_nullable?: false}},
protocol_name: {:string, %{is_nullable?: false}},
leader: {:string, %{is_nullable?: false}},
member_id: {:string, %{is_nullable?: false}},
members:
{{:array,
[
member_id: {:string, %{is_nullable?: false}},
metadata: {:bytes, %{is_nullable?: false}}
]}, %{is_nullable?: false}}
]
def response_schema(4),
do: [
throttle_time_ms: {:int32, %{is_nullable?: false}},
error_code: {:int16, %{is_nullable?: false}},
generation_id: {:int32, %{is_nullable?: false}},
protocol_name: {:string, %{is_nullable?: false}},
leader: {:string, %{is_nullable?: false}},
member_id: {:string, %{is_nullable?: false}},
members:
{{:array,
[
member_id: {:string, %{is_nullable?: false}},
metadata: {:bytes, %{is_nullable?: false}}
]}, %{is_nullable?: false}}
]
def response_schema(5),
do: [
throttle_time_ms: {:int32, %{is_nullable?: false}},
error_code: {:int16, %{is_nullable?: false}},
generation_id: {:int32, %{is_nullable?: false}},
protocol_name: {:string, %{is_nullable?: false}},
leader: {:string, %{is_nullable?: false}},
member_id: {:string, %{is_nullable?: false}},
members:
{{:array,
[
member_id: {:string, %{is_nullable?: false}},
group_instance_id: {:string, %{is_nullable?: true}},
metadata: {:bytes, %{is_nullable?: false}}
]}, %{is_nullable?: false}}
]
def response_schema(6),
do: [
throttle_time_ms: {:int32, %{is_nullable?: false}},
error_code: {:int16, %{is_nullable?: false}},
generation_id: {:int32, %{is_nullable?: false}},
protocol_name: {:compact_string, %{is_nullable?: false}},
leader: {:compact_string, %{is_nullable?: false}},
member_id: {:compact_string, %{is_nullable?: false}},
members:
{{:compact_array,
[
member_id: {:compact_string, %{is_nullable?: false}},
group_instance_id: {:compact_string, %{is_nullable?: true}},
metadata: {:compact_bytes, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, %{}}
]}, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, %{}}
]
def response_schema(7),
do: [
throttle_time_ms: {:int32, %{is_nullable?: false}},
error_code: {:int16, %{is_nullable?: false}},
generation_id: {:int32, %{is_nullable?: false}},
protocol_type: {:compact_string, %{is_nullable?: true}},
protocol_name: {:compact_string, %{is_nullable?: true}},
leader: {:compact_string, %{is_nullable?: false}},
member_id: {:compact_string, %{is_nullable?: false}},
members:
{{:compact_array,
[
member_id: {:compact_string, %{is_nullable?: false}},
group_instance_id: {:compact_string, %{is_nullable?: true}},
metadata: {:compact_bytes, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, %{}}
]}, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, %{}}
]
def response_schema(8),
do: [
throttle_time_ms: {:int32, %{is_nullable?: false}},
error_code: {:int16, %{is_nullable?: false}},
generation_id: {:int32, %{is_nullable?: false}},
protocol_type: {:compact_string, %{is_nullable?: true}},
protocol_name: {:compact_string, %{is_nullable?: true}},
leader: {:compact_string, %{is_nullable?: false}},
member_id: {:compact_string, %{is_nullable?: false}},
members:
{{:compact_array,
[
member_id: {:compact_string, %{is_nullable?: false}},
group_instance_id: {:compact_string, %{is_nullable?: true}},
metadata: {:compact_bytes, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, %{}}
]}, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, %{}}
]
def response_schema(9),
do: [
throttle_time_ms: {:int32, %{is_nullable?: false}},
error_code: {:int16, %{is_nullable?: false}},
generation_id: {:int32, %{is_nullable?: false}},
protocol_type: {:compact_string, %{is_nullable?: true}},
protocol_name: {:compact_string, %{is_nullable?: true}},
leader: {:compact_string, %{is_nullable?: false}},
skip_assignment: {:boolean, %{is_nullable?: false}},
member_id: {:compact_string, %{is_nullable?: false}},
members:
{{:compact_array,
[
member_id: {:compact_string, %{is_nullable?: false}},
group_instance_id: {:compact_string, %{is_nullable?: true}},
metadata: {:compact_bytes, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, %{}}
]}, %{is_nullable?: false}},
tag_buffer: {:tag_buffer, %{}}
]
def response_schema(unkown_version),
do: raise("Unknown version #{unkown_version} for message JoinGroup")
end