Module: Karafka::Pro::Processing::ConsumerGroups::Strategies::Vp::Default

Includes:
Default
Included in:
Aj::DlqMomVp, Aj::FtrLrjMomVp, Aj::LrjMomVp, Aj::MomVp, Dlq::FtrLrjMomVp, Dlq::FtrLrjVp, Dlq::FtrVp, Dlq::LrjVp, Dlq::Vp, Ftr::Vp, Lrj::FtrVp, Lrj::Vp, Mom::Vp
Defined in:
lib/karafka/pro/processing/consumer_groups/strategies/vp/default.rb

Overview

Just Virtual Partitions enabled

Constant Summary collapse

FEATURES =

Features for this strategy

%i[
  virtual_partitions
].freeze

Instance Method Summary collapse

Methods included from Default

#handle_after_consume, #handle_before_consume, #handle_before_schedule_tick, #handle_consume, #handle_revoked, #handle_tick, #mark_in_memory, #mark_with_transaction, #store_offset_metadata, #transaction

Methods included from Karafka::Processing::ConsumerGroups::Strategies::Default

#commit_offsets, #commit_offsets!, #handle_after_consume, #handle_before_consume, #handle_consume, #handle_eofed, #handle_idle, #handle_initialized, #handle_revoked, #handle_shutdown, #handle_wrap

Methods included from Karafka::Processing::ConsumerGroups::Strategies::Base

#handle_after_consume, #handle_before_consume, #handle_consume, #handle_idle, #handle_revoked, #handle_shutdown

Instance Method Details

#collapse_until!(offset) ⇒ Object

Note:

Keep in mind, that if a batch contains this but also messages earlier messages that should be collapsed, all will continue to operate in a collapsed mode until first full batch with only messages that should not be collapsed.

Parameters:

  • offset (Integer) —

    first offset from which we should not operate in a collapsed mode.



148
149
150
 
# File 'lib/karafka/pro/processing/consumer_groups/strategies/vp/default.rb', line 148

def collapse_until!(offset)
  coordinator.collapse_until!(offset)
end

#collapsed? ⇒ Boolean

Returns is the virtual processing collapsed in the context of given consumer.

Returns:

  • (Boolean) —

    is the virtual processing collapsed in the context of given consumer.



139
140
141
 
# File 'lib/karafka/pro/processing/consumer_groups/strategies/vp/default.rb', line 139

def collapsed?
  coordinator.collapsed?
end

#failing? ⇒ Boolean

Note:

We've named it #failing? instead of #failure? because it aims to be used from within virtual partitions where we want to have notion of collective failing not just "local" to our processing. We "are" failing with other virtual partitions raising an error, but locally we are still processing.

Returns true if any of virtual partition we're operating in the entangled mode has already failed and we know we are failing collectively. Useful for early stop to minimize number of things processed twice.

Returns:

  • (Boolean) —

    true if any of virtual partition we're operating in the entangled mode has already failed and we know we are failing collectively. Useful for early stop to minimize number of things processed twice.



160
161
162
 
# File 'lib/karafka/pro/processing/consumer_groups/strategies/vp/default.rb', line 160

def failing?
  coordinator.failure?
end

#mark_as_consumed(message, offset_metadata = @_current_offset_metadata) ⇒ Object

Note:

This virtual offset management uses a regular default marking API underneath. We do not alter the "real" marking API, as VPs are just one of many cases we want to support and we do not want to impact them with collective offsets management

Parameters:

  • message (Karafka::Messages::Message) —

    marks message as consumed

  • offset_metadata (String, nil) (defaults to: @_current_offset_metadata) 


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
 
# File 'lib/karafka/pro/processing/consumer_groups/strategies/vp/default.rb', line 54

def mark_as_consumed(message, offset_metadata = @_current_offset_metadata)
  if @_in_transaction && !collapsed?
    mark_in_transaction(message, offset_metadata, true)
  elsif collapsed?
    super
  else
    manager = coordinator.virtual_offset_manager

    coordinator.synchronize do
      manager.mark(message, offset_metadata)
      # If this is last marking on a finished flow, we can use the original
      # last message and in order to do so, we need to mark all previous messages as
      # consumed as otherwise the computed offset could be different
      # We mark until our offset just in case of a DLQ flow or similar, where we do not
      # want to mark all but until the expected location
      manager.mark_until(message, offset_metadata) if coordinator.finished?

      # When there is no continuous stream of virtually marked offsets from the
      # lowest registered one, there is no real offset to commit yet. This is not
      # a failure, hence we need to report ownership truthfully: true as long as
      # the assignment is still owned
      return !revoked? unless manager.markable?

      super(*manager.markable)
    end
  end
ensure
  @_current_offset_metadata = nil
end

#mark_as_consumed!(message, offset_metadata = @_current_offset_metadata) ⇒ Object

Parameters:

  • message (Karafka::Messages::Message) —

    blocking marks message as consumed

  • offset_metadata (String, nil) (defaults to: @_current_offset_metadata) 


86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/karafka/pro/processing/consumer_groups/strategies/vp/default.rb', line 86

def mark_as_consumed!(message, offset_metadata = @_current_offset_metadata)
  if @_in_transaction && !collapsed?
    mark_in_transaction(message, offset_metadata, false)
  elsif collapsed?
    super
  else
    manager = coordinator.virtual_offset_manager

    coordinator.synchronize do
      manager.mark(message, offset_metadata)
      manager.mark_until(message, offset_metadata) if coordinator.finished?

      # No real offset to commit yet is not a failure, see `#mark_as_consumed`
      return !revoked? unless manager.markable?

      super(*manager.markable)
    end
  end
ensure
  @_current_offset_metadata = nil
end

#mark_in_transaction(message, offset_metadata, async) ⇒ Object

Stores the next offset for processing inside of the transaction when collapsed and accumulates marking as consumed in the local buffer.

Due to nature of VPs we cannot provide full EOS support but we can simulate it, making sure that no offset are stored unless transaction is finished. We do it by accumulating the post-transaction marking requests and after it is successfully done we mark each as consumed. This effectively on errors "rollbacks" the state and prevents offset storage.

Since the EOS here is "weak", we do not have to worry about the race-conditions and we do not have to have any mutexes.

Parameters:

  • message (Messages::Message) —

    message we want to commit inside of a transaction

  • offset_metadata (String, nil) —

    offset metadata or nil if none

  • async (Boolean) —

    should we mark in async or sync way (applicable only to post transaction state synchronization usage as within transaction it is always sync)

Raises:



124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/karafka/pro/processing/consumer_groups/strategies/vp/default.rb', line 124

def mark_in_transaction(message, offset_metadata, async)
  raise Errors::TransactionRequiredError unless @_in_transaction
  # Prevent from attempts of offset storage when we no longer own the assignment
  raise Errors::AssignmentLostError if revoked?

  return super if collapsed?
  # If this is user post-execution transaction (one initiated by the system) we should
  # delegate to the original implementation that will store the offset via the producer
  return super if @_transaction_internal

  @_transaction_marked << [message, offset_metadata, async]
end

#synchronize ⇒ Object

Allows for cross-virtual-partition consumers locks

This is not needed in the non-VP flows except LRJ because there is always only one consumer per partition at the same time, so no coordination is needed directly for the end users. With LRJ it is needed and provided in the LRJ::Default strategy, because lifecycle events on revocation can run in parallel to the LRJ job as it is non-blocking.



171
172
173
 
# File 'lib/karafka/pro/processing/consumer_groups/strategies/vp/default.rb', line 171

def synchronize(&)
  coordinator.shared_mutex.synchronize(&)
end