Class: Rdkafka::Callbacks::IncrementalAlterConfigsHandler

Inherits:
BaseHandler
  • Object
show all
Defined in:
lib/rdkafka/callbacks/incremental_alter_configs_handler.rb

Overview

Handles RD_KAFKA_EVENT_INCREMENTALALTERCONFIGS_RESULT events

Class Method Summary collapse

Class Method Details

.call(event_ptr) ⇒ void

This method returns an undefined value.

Resolves the incremental-alter-configs handle from its result event

Parameters:

  • event_ptr (FFI::Pointer) —

    pointer to the event



12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/rdkafka/callbacks/incremental_alter_configs_handler.rb', line 12

def call(event_ptr)
  incremental_alter = IncrementalAlterConfigsResult.new(event_ptr)
  incremental_alter_handle_ptr = Rdkafka::Bindings.rd_kafka_event_opaque(event_ptr)

  if incremental_alter_handle = Rdkafka::Admin::IncrementalAlterConfigsHandle.remove(incremental_alter_handle_ptr.address)
    incremental_alter_handle[:response] = incremental_alter.result_error
    incremental_alter_handle.broker_message = read_event_string(incremental_alter.error_string)

    # Parsing can raise on per-resource errors. The exception is captured and re-raised on
    # the waiting thread, since an exception leaving this callback would unwind through
    # librdkafka native frames
    incremental_alter_handle.result = begin
      if incremental_alter.result_error == Rdkafka::Bindings::RD_KAFKA_RESP_ERR_NO_ERROR
        Rdkafka::Admin::IncrementalAlterConfigsReport.new(
          config_entries: incremental_alter.results,
          entry_count: incremental_alter.results_count
        )
      else
        Rdkafka::Admin::IncrementalAlterConfigsReport.new(config_entries: FFI::Pointer::NULL, entry_count: 0)
      end
    rescue => e
      e
    end

    incremental_alter_handle.unlock
  end
end