Class: Karafka::Routing::SubscriptionGroup

Inherits:
Object
  • Object
show all
Defined in:
lib/karafka/routing/subscription_group.rb

Overview

Note:

One subscription group will always belong to one consumer group, but one consumer group can have multiple subscription groups.

Object representing a set of single consumer group topics that can be subscribed together with one connection.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(position, topics) ⇒ SubscriptionGroup 



50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/karafka/routing/subscription_group.rb', line 50

def initialize(position, topics)
  @details = topics.first.subscription_group_details
  @name = @details.fetch(:name)
  @group = topics.first.group
  # We include the owning group id here because we want to have unique ids of subscription
  # groups across the system. Otherwise user could set the same name for multiple
  # subscription groups in many groups effectively having same id for different entities
  @id = "#{@group.id}_#{@name}_#{position}"
  @position = position
  @topics = topics
  @kafka = build_kafka
end

Instance Attribute Details

#group ⇒ Object (readonly) Also known as: consumer_group

Returns the value of attribute group.



17
18
19
 
# File 'lib/karafka/routing/subscription_group.rb', line 17

def group
  @group
end

#id ⇒ Object (readonly)

Returns the value of attribute id.



17
18
19
 
# File 'lib/karafka/routing/subscription_group.rb', line 17

def id
  @id
end

#kafka ⇒ Object (readonly)

Returns the value of attribute kafka.



17
18
19
 
# File 'lib/karafka/routing/subscription_group.rb', line 17

def kafka
  @kafka
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



17
18
19
 
# File 'lib/karafka/routing/subscription_group.rb', line 17

def name
  @name
end

#position ⇒ Object (readonly)

Returns the value of attribute position.



17
18
19
 
# File 'lib/karafka/routing/subscription_group.rb', line 17

def position
  @position
end

#topics ⇒ Object (readonly)

Returns the value of attribute topics.



17
18
19
 
# File 'lib/karafka/routing/subscription_group.rb', line 17

def topics
  @topics
end

Class Method Details

.id ⇒ String

Generates new subscription group id that will be used in case of anonymous subscription groups



33
34
35
36
37
38
39
40
41
42
 
# File 'lib/karafka/routing/subscription_group.rb', line 33

def id
  ID_MUTEX.synchronize do
    @group_counter ||= 0
    @group_counter += 1

    Digest::SHA256.hexdigest(
      @group_counter.to_s
    )[0..11]
  end
end

Instance Method Details

#active? ⇒ Boolean 



84
85
86
 
# File 'lib/karafka/routing/subscription_group.rb', line 84

def active?
  activity_manager.active?(:subscription_groups, name)
end

#assignments(_consumer) ⇒ false, Rdkafka::Consumer::TopicPartitionList 



101
102
103
 
# File 'lib/karafka/routing/subscription_group.rb', line 101

def assignments(_consumer)
  false
end

#group_id ⇒ String Also known as: consumer_group_id 



65
66
67
 
# File 'lib/karafka/routing/subscription_group.rb', line 65

def group_id
  kafka[:"group.id"]
end

#max_messages ⇒ Integer 



74
75
76
 
# File 'lib/karafka/routing/subscription_group.rb', line 74

def max_messages
  @topics.first.max_messages
end

#max_wait_time ⇒ Integer 



79
80
81
 
# File 'lib/karafka/routing/subscription_group.rb', line 79

def max_wait_time
  @topics.first.max_wait_time
end

#refresh ⇒ Object

Refreshes the configuration of this subscription group if needed based on the execution context.

Since the initial routing setup happens in the supervisor, it is inherited by the children. This causes incomplete assignment of group.instance.id which is not expanded with proper node identifier. This refreshes this if needed when in swarm.



117
118
119
120
121
122
 
# File 'lib/karafka/routing/subscription_group.rb', line 117

def refresh
  return unless node
  return unless kafka.key?(:"group.instance.id")

  @kafka = build_kafka
end

#subscriptions ⇒ false, Array<String>

Note:

Most of the time it should not include inactive topics but in case of pattern matching the matcher topics become inactive down the road, hence we filter out so they are later removed.

Returns names of topics to which we should subscribe or false when operating only on direct assignments.



94
95
96
 
# File 'lib/karafka/routing/subscription_group.rb', line 94

def subscriptions
  topics.select(&:active?).map(&:subscription_name)
end

#to_s ⇒ String

Note:

This is an alias for displaying in places where we print the stringified version.

Returns id of the subscription group.



107
108
109
 
# File 'lib/karafka/routing/subscription_group.rb', line 107

def to_s
  id
end