Module: Karafka::Pro::Routing::Features::ConsumerGroups::ScheduledMessages::Builder

Included in:
Proxy
Defined in:
lib/karafka/pro/routing/features/consumer_groups/scheduled_messages/builder.rb

Overview

Routing extensions for scheduled messages

Instance Method Summary collapse

Instance Method Details

#scheduled_messages(topic_name = false, &block) ⇒ Object

Note:

Namespace for topics should include the divider as it is not automatically added.

Enabled scheduled messages operations and adds needed topics and other stuff.

Parameters:

  • topic_name (String, false) (defaults to: false) —

    name for scheduled messages topic that is also used as a group identifier. Users can have multiple schedule topics flows to prevent key collisions, prioritize and do other stuff. false if not active.

  • block (Proc) —

    optional reconfiguration of the topics definitions.



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/karafka/pro/routing/features/consumer_groups/scheduled_messages/builder.rb', line 47

def scheduled_messages(topic_name = false, &block)
  return unless topic_name

  # Load zlib only if user enables scheduled messages
  require "zlib"

  # We set it to 5 so we have enough space to handle more events. All related topics
  # should have same partition count.
  default_partitions = 5
  msg_cfg = App.config.scheduled_messages
  states_topic_name = "#{topic_name}#{msg_cfg.states_postfix}"

  # Declarative topic definitions live independently of routing. We declare the
  # broker-side structure here so it is managed by the CLI topics commands without
  # relying on the (to be retired) routing `config(...)` bridge.
  #
  # We guard each declaration with `find_topic` because `Declaratives::Builder#topic`
  # re-applies its block on an already existing declaration (additive semantics),
  # unlike the old bridge which used `||=` and kept the first declaration. Without the
  # guard, our framework defaults would clobber a topic the user may have declared
  # themselves via `App.declaratives.draw`. Skipping when already present preserves
  # first-declaration-wins.
  App.declaratives.draw do
    # This is a setup that should allow messages to be compacted fairly fast. Since
    # each dispatched message should be removed via tombstone, they do not have to
    # be present in the topic for too long.
    unless find_topic(topic_name)
      topic(topic_name) do
        partitions default_partitions
        config(
          # Will ensure, that after tombstone is present, given scheduled message,
          # that has been dispatched is removed by Kafka
          "cleanup.policy": "compact",
          # When 10% or more dispatches are done, compact data
          "min.cleanable.dirty.ratio": 0.1,
          # Frequent segment rotation to support intense compaction
          "segment.ms": 3_600_000,
          "delete.retention.ms": 3_600_000,
          "segment.bytes": 52_428_800
        )
      end
    end

    # Holds states of scheduler per each of the partitions. Same partition count as
    # the messages topic since they tick independently per partition.
    unless find_topic(states_topic_name)
      topic(states_topic_name) do
        partitions default_partitions
        config(
          "cleanup.policy": "compact",
          "min.cleanable.dirty.ratio": 0.1,
          "segment.ms": 3_600_000,
          "delete.retention.ms": 3_600_000,
          "segment.bytes": 52_428_800
        )
      end
    end
  end

  consumer_group msg_cfg.group_id do
    # Registers the primary topic that we use to control schedules execution. This is
    # the one that we use to trigger scheduled messages.
    messages_topic = topic(topic_name) do
      instance_eval(&block) if block && block.arity.zero?

      consumer msg_cfg.consumer_class

      deserializers(
        headers: msg_cfg.deserializers.headers
      )

      # Because the topic method name as well as builder proxy method name is the same
      # we need to reference it via target directly
      target.scheduled_messages(true)

      # We manage offsets directly because messages can have both schedules and
      # commands and we need to apply them only when we need to
      manual_offset_management(true)

      # We use multi-batch operations and in-memory state for schedules. This needs to
      # always operate without re-creation.
      consumer_persistence(true)

      # This needs to be enabled for the eof to work correctly
      kafka(
        "enable.partition.eof": true,
        "auto.offset.reset": "earliest",
        inherit: true
      )
      eofed(true)

      # Since this is a topic that gets replayed because of schedule management, we do
      # want to get more data faster during recovery
      max_messages(10_000)

      max_wait_time(1_000)

      # This is the core of execution. Since we dispatch data in time intervals, we
      # need to be able to do this even when no new data is coming
      periodic(
        interval: msg_cfg.interval,
        during_pause: false,
        during_retry: false
      )

      # If this is the direct schedules redefinition style, we run it
      # The second one (see end of this method) allows for linear reconfiguration of
      # both the topics
      instance_eval(&block) if block && block.arity.zero?
    end

    # Holds states of scheduler per each of the partitions since they tick
    # independently. We only hold future statistics not to have to deal with
    # any type of state restoration
    states_topic = topic(states_topic_name) do
      active(false)
      target.scheduled_messages(true)
      deserializers(
        payload: msg_cfg.deserializers.payload
      )
    end

    yield(messages_topic, states_topic) if block && block.arity.positive?
  end
end