Class: Kafka::Protocol::OffsetFetchResponse

18
19
20

# File 'lib/kafka/protocol/offset_fetch_response.rb', line 18
def initialize(topics:)
 @topics = topics
end

16
17
18

# File 'lib/kafka/protocol/offset_fetch_response.rb', line 16
def topics
 @topics
end

33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

# File 'lib/kafka/protocol/offset_fetch_response.rb', line 33
def self.decode(decoder)
 topics = decoder.array {
 topic = decoder.string
 partitions = decoder.array {
 partition = decoder.int32
 info = PartitionOffsetInfo .new (
 offset: decoder.int64,
 metadata: decoder.string,
 error_code: decoder.int16,
 )
 [partition, info]
 }
 [topic, Hash[partitions]]
 }
 new(topics: Hash[topics])
end

22
23
24
25
26
27
28
29
30
31

# File 'lib/kafka/protocol/offset_fetch_response.rb', line 22
def offset_for(topic, partition)
 offset_info = topics.fetch(topic).fetch(partition, nil)
 if offset_info
 Protocol .handle_error (offset_info.error_code)
 offset_info.offset
 else
 -1
 end
end