Class: Karafka::Pro::Processing::VirtualPartitions::Distributors::Consistent

Inherits:

Base

• Object
• Base
• Karafka::Pro::Processing::VirtualPartitions::Distributors::Consistent

Defined in:

lib/karafka/pro/processing/virtual_partitions/distributors/consistent.rb

Overview

Consistent distributor that ensures messages with the same partition key are always processed in the same virtual partition

Instance Method Summary

• #call(messages) ⇒ Hash<Integer, Array<Karafka::Messages::Message>>

  Hash with group ids as keys and message groups as values.

Methods inherited from Base

#initialize

Constructor Details

This class inherits a constructor from Karafka::Pro::Processing::VirtualPartitions::Distributors::Base

Instance Method Details

#call(messages) ⇒ Hash<Integer, Array<Karafka::Messages::Message>>

Returns hash with group ids as keys and message groups as values.

Parameters:

• messages (Array<Karafka::Messages::Message>) —

  messages to distribute

Returns:

• (Hash<Integer, Array<Karafka::Messages::Message>>) —

  hash with group ids as keys and message groups as values

# File 'lib/karafka/pro/processing/virtual_partitions/distributors/consistent.rb', line 17

def call(messages)
  messages
    .group_by { |msg| config.reducer.call(config.partitioner.call(msg)) }
    .to_h
end