Class: Kafka::Compressor

Inherits:
Object
  • Object
show all
Defined in:
lib/kafka/compressor.rb

Overview

Compresses message sets using a specified codec.

A message set is only compressed if its size meets the defined threshold.

Instrumentation

Whenever a message set is compressed, the notification compress.compressor.kafka will be emitted with the following payload:

  • message_count – the number of messages in the message set.
  • uncompressed_bytesize – the byte size of the original data.
  • compressed_bytesize – the byte size of the compressed data.

Instance Method Summary collapse

Constructor Details

#initialize(codec_name:, threshold:) ⇒ Compressor

Returns a new instance of Compressor.

Parameters:

  • codec_name (Symbol, nil)
  • threshold (Integer) —

    the minimum number of messages in a message set that will trigger compression.



23
24
25
26
 
# File 'lib/kafka/compressor.rb', line 23

def initialize(codec_name:, threshold:)
  @codec = Compression.find_codec(codec_name)
  @threshold = threshold
end

Instance Method Details

#compress(message_set) ⇒ Protocol::MessageSet

Parameters:

Returns:



30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/kafka/compressor.rb', line 30

def compress(message_set)
  return message_set if @codec.nil? || message_set.size < @threshold

  compressed_data = compress_data(message_set)

  wrapper_message = Protocol::Message.new(
    value: compressed_data,
    codec_id: @codec.codec_id,
  )

  Protocol::MessageSet.new(messages: [wrapper_message])
end