Class: Journaled::KinesisSequentialSender

Inherits:
Object
  • Object
show all
Defined in:
lib/journaled/kinesis_sequential_sender.rb

Overview

Sends batches of events to Kinesis using the PutRecord single-event API

This class handles:

  • Sending events individually in order to support guaranteed ordering

  • Stopping on first transient failure to preserve ordering

  • Classifying errors as transient vs permanent

Returns structured results for the caller to handle event state management.

Constant Summary collapse

PERMANENT_ERROR_CLASSES = 
[
  Aws::Kinesis::Errors::ValidationException,
].freeze

Instance Method Summary collapse

Instance Method Details

#send_batch(events) ⇒ Hash

Send a batch of database events to Kinesis

Sends events one at a time to guarantee ordering. Stops on first transient failure.

Parameters:

Returns:

  • (Hash) —

    Result with:

    • succeeded: Array of successfully sent events

    • failed: Array of FailedEvent structs (only permanent failures)



25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/journaled/kinesis_sequential_sender.rb', line 25

def send_batch(events)
  result = { succeeded: [], failed: [] }

  events.each do |event|
    event_result = send_event(event)
    if event_result.is_a?(Journaled::KinesisFailedEvent)
      if event_result.transient?
        emit_transient_failure_metric
        break
      else
        result[:failed] << event_result
      end
    else
      result[:succeeded] << event_result
    end
  end

  result
end