kafkaex.ex

KafkaEx.create_worker(:foo, uris: Application.get_env(:kafka_ex, :brokers), consumer_group: "bar")
Enum.each(1..1000, fn x -> KafkaEx.produce("foo", 0, "bar#{x}") end) 
KafkaEx.fetch("foo", 0, worker_name: :foo, max_bytes: 100)   
# output
"""
[%KafkaEx.Protocol.Fetch.Response{partitions: [%{error_code: 0,
     hw_mark_offset: 1002, last_offset: 2,
     message_set: [%KafkaEx.Protocol.Fetch.Message{attributes: 0,
       crc: 2491399380, key: "", offset: 0, value: "bar"},
      %KafkaEx.Protocol.Fetch.Message{attributes: 0, crc: 702483308, key: "",
       offset: 1, value: "bar2"},
      %KafkaEx.Protocol.Fetch.Message{attributes: 0, crc: 2966837974, key: "",
       offset: 2, value: "bar1"}], partition: 0}], topic: "foo"}]
"""

KafkaEx.fetch("foo", 0, worker_name: :foo, max_bytes: 100)
# output
"""
[%KafkaEx.Protocol.Fetch.Response{partitions: [%{error_code: 0,
     hw_mark_offset: 1002, last_offset: 5,
     message_set: [%KafkaEx.Protocol.Fetch.Message{attributes: 0,
       crc: 702483308, key: "", offset: 3, value: "bar2"},
      %KafkaEx.Protocol.Fetch.Message{attributes: 0, crc: 1591229434, key: "",
       offset: 4, value: "bar3"},
      %KafkaEx.Protocol.Fetch.Message{attributes: 0, crc: 3233590873, key: "",
       offset: 5, value: "bar4"}], partition: 0}], topic: "foo"}]
"""

KafkaEx.fetch("foo", 0, worker_name: :foo, max_bytes: 100)

# output
"""
[%KafkaEx.Protocol.Fetch.Response{partitions: [%{error_code: 0,
     hw_mark_offset: 1002, last_offset: 8,
     message_set: [%KafkaEx.Protocol.Fetch.Message{attributes: 0,
       crc: 3082526415, key: "", offset: 6, value: "bar5"},
      %KafkaEx.Protocol.Fetch.Message{attributes: 0, crc: 783470453, key: "",
       offset: 7, value: "bar6"},
      %KafkaEx.Protocol.Fetch.Message{attributes: 0, crc: 1505099747, key: "",
       offset: 8, value: "bar7"}], partition: 0}], topic: "foo"}]
"""