Class: Karafka::Pro::ScheduledMessages::Dispatcher

Inherits:
Object
  • Object
show all
Defined in:
lib/karafka/pro/scheduled_messages/dispatcher.rb

Overview

Dispatcher responsible for dispatching the messages to appropriate target topics and for dispatching other messages. All messages (aside from the once users dispatch with the envelope) are sent via this dispatcher.

Messages are buffered and dispatched in batches to improve dispatch performance.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(topic, partition) ⇒ Dispatcher

Returns a new instance of Dispatcher.

Parameters:

  • topic (String) —

    consumed topic name

  • partition (Integer) —

    consumed partition



45
46
47
48
49
50
51
52
53
54
 
# File 'lib/karafka/pro/scheduled_messages/dispatcher.rb', line 45

def initialize(topic, partition)
  @topic = topic
  @partition = partition
  @buffer = []
  # Source (daily buffer) key aligned 1:1 with each `@buffer` entry, so `#flush` can report
  # which keys were confirmed delivered per chunk and the consumer can evict them
  # incrementally instead of only after the whole flush succeeds
  @keys = []
  @serializer = Serializer.new
end

Instance Attribute Details

#buffer ⇒ Array<Hash> (readonly)

Returns buffer with message hashes for dispatch.

Returns:

  • (Array<Hash>) —

    buffer with message hashes for dispatch



41
42
43
 
# File 'lib/karafka/pro/scheduled_messages/dispatcher.rb', line 41

def buffer
  @buffer
end

Instance Method Details

#<<(message) ⇒ Object

Note:

This method adds the message to the buffer, does not dispatch it.

Note:

It also produces needed tombstone event as well as an audit log message

Prepares the scheduled message to the dispatch to the target topic. Extracts all the "schedule_" details and prepares it, so the dispatched message goes with the expected attributes to the desired location. Alongside of that it actually builds 2 (1 if logs off) messages: tombstone event matching the schedule so it is no longer valid and the log message that has the same data as the dispatched message. Helpful when debugging.

Parameters:



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/karafka/pro/scheduled_messages/dispatcher.rb', line 67

def <<(message)
  target_headers = message.raw_headers.merge(
    "schedule_source_topic" => @topic,
    "schedule_source_partition" => @partition.to_s,
    "schedule_source_offset" => message.offset.to_s,
    "schedule_source_key" => message.key
  ).compact

  target = {
    payload: message.raw_payload,
    headers: target_headers
  }

  extract(target, message.headers, :topic)
  extract(target, message.headers, :partition)
  extract(target, message.headers, :key)
  extract(target, message.headers, :partition_key)

  # `message.key` is pushed to `@keys` once per `@buffer` entry (here and below), not once
  # per message: `@keys` must stay 1:1 aligned with `@buffer` (see `#flush`) so that
  # shifting a chunk off both arrays together always yields the correct keys for that
  # chunk. The target and its tombstone are two separate `@buffer` entries for the same
  # schedule, so the same key is intentionally paired with each. `DailyBuffer#delete` is a
  # plain `Hash#delete`, so a key evicted twice within the same yielded chunk (once per
  # entry) is a harmless no-op the second time.
  @buffer << target
  @keys << message.key

  # Tombstone message so this schedule is no longer in use and gets removed from Kafka by
  # Kafka itself during compacting. It will not cancel it because already dispatched but
  # will cause it not to be sent again and will be marked as dispatched.
  @buffer << Proxy.tombstone(message: message)
  @keys << message.key
end

#flush {|keys| ... } ⇒ Object

Sends all messages to Kafka in a sync way. We use sync with batches to prevent overloading. When transactional producer in use, this will be wrapped in a transaction automatically.

Yield Parameters:

  • keys (Array<String>) —

    of the chunk that was just confirmed delivered. Yielded after each chunk's sync produce returns, so the caller can evict those keys from the daily buffer incrementally. If a later chunk raises, the chunks already produced have still been reported, so a non-transactional producer will not re-dispatch them.

Raises:

  • (ArgumentError) —

    when called without a block, since a caller that cannot observe per-chunk confirmations cannot evict incrementally and would reintroduce the whole-flush-or-nothing duplicate-dispatch window this method exists to close



129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/karafka/pro/scheduled_messages/dispatcher.rb', line 129

def flush
  raise ArgumentError, "#flush requires a block to report per-chunk confirmations" unless block_given?

  until @buffer.empty?
    batch_size = config.flush_batch_size

    # A message's target and its tombstone are always buffered as an adjacent pair (see
    # `#<<`). Rounding the chunk size up to even guarantees a chunk boundary can never
    # fall between them - with an odd (or 1) `flush_batch_size`, a chunk could otherwise
    # confirm and yield a key whose target was produced but whose tombstone was not, and a
    # later chunk failing would then leave that schedule non-tombstoned in Kafka, to be
    # re-dispatched after a restart/reload.
    batch_size += 1 if batch_size.odd?

    messages = @buffer.shift(batch_size)
    keys = @keys.shift(batch_size)

    config.producer.produce_many_sync(messages)

    yield(keys)
  end
ensure
  # Whether flush finished normally (buffer already empty here, so this is a no-op) or
  # raised partway through, drop anything left. Those messages are still in the daily
  # buffer (their keys were never yielded, so never evicted) and will be re-buffered on
  # the next tick, so stale leftovers here must not be dispatched a second time.
  @buffer.clear
  @keys.clear
end

#state(tracker) ⇒ Object

Note:

This is dispatched async because it's just a statistical metric.

Builds and dispatches the state report message with schedules details

Parameters:



107
108
109
110
111
112
113
114
115
116
 
# File 'lib/karafka/pro/scheduled_messages/dispatcher.rb', line 107

def state(tracker)
  config.producer.produce_async(
    topic: "#{@topic}#{config.states_postfix}",
    payload: @serializer.state(tracker),
    # We use the state as a key, so we always have one state transition data available
    key: "#{tracker.state}_state",
    partition: @partition,
    headers: { "zlib" => "true" }
  )
end