Class: Karafka::Web::Pro::Ui::Controllers::Consumers::Partitions::OffsetsController

Inherits:
BaseController show all
Defined in:
lib/karafka/web/pro/ui/controllers/consumers/partitions/offsets_controller.rb

Overview

Partition offset management controller at the consumer group level

Instance Method Summary collapse

Methods inherited from ConsumersController

#details, #index, #performance, #subscriptions

Instance Method Details

#edit(consumer_group_id, topic, partition_id) ⇒ Object

Displays the offset edit page with the edit form or a warning when not applicable

Parameters:

  • consumer_group_id (String)
  • topic (String)
  • partition_id (Integer) 


47
48
49
50
51
 
# File 'lib/karafka/web/pro/ui/controllers/consumers/partitions/offsets_controller.rb', line 47

def edit(consumer_group_id, topic, partition_id)
  bootstrap!(consumer_group_id, topic, partition_id)

  render
end

#update(consumer_group_id, topic, partition_id) ⇒ Object

Triggers the offset change in the running process via the commanding API

Parameters:

  • consumer_group_id (String)
  • topic (String)
  • partition_id (Integer) 


58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/karafka/web/pro/ui/controllers/consumers/partitions/offsets_controller.rb', line 58

def update(consumer_group_id, topic, partition_id)
  edit(consumer_group_id, topic, partition_id)

  offset = params.int(:offset)
  prevent_overtaking = params.bool(:prevent_overtaking)
  force_resume = params.bool(:force_resume)

  # Broadcast to all processes with matchers to filter by consumer group,
  # topic, and partition
  Commanding::Dispatcher.request(
    Commanding::Commands::Partitions::Seek.name,
    {
      consumer_group_id: consumer_group_id,
      topic: topic,
      partition_id: partition_id,
      offset: offset,
      prevent_overtaking: prevent_overtaking,
      force_resume: force_resume
    },
    matchers: {
      consumer_group_id: consumer_group_id,
      topic: topic,
      partition_id: partition_id
    }
  )

  redirect(
    :previous,
    success: format_flash(
      "Initiated offset adjustment to ? for ?#? in consumer group ?",
      offset,
      topic,
      partition_id,
      consumer_group_id
    )
  )
end