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>}

  Distributes messages ensuring consistent routing where messages with the same partition key always go to the same virtual partition.

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>}

Distributes messages ensuring consistent routing where messages with the same partition key always go to the same virtual partition

Parameters:

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

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 19

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