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

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

Overview

Controller for managing partition pauses at the consumer group level.

Constant Summary

Constants inherited from Ui::Controllers::BaseController

Ui::Controllers::BaseController::Models

Instance Attribute Summary

Attributes inherited from Ui::Controllers::BaseController

#params, #session

Instance Method Summary collapse

Methods inherited from ConsumersController

#details, #index, #performance, #subscriptions

Methods inherited from Ui::Controllers::BaseController

#cache, #initialize

Methods included from Ui::Controllers::Requests::Hookable

included, #run_after_hooks, #run_before_hooks

Constructor Details

This class inherits a constructor from Karafka::Web::Ui::Controllers::BaseController

Instance Method Details

#create(consumer_group_id, topic, partition_id) ⇒ Object

Dispatches the partition pause command to all processes

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
 
# File 'lib/karafka/web/pro/ui/controllers/consumers/partitions/pauses_controller.rb', line 58

def create(consumer_group_id, topic, partition_id)
  new(consumer_group_id, topic, partition_id)

  # Broadcast to all processes with matchers to filter by consumer group,
  # topic, and partition
  Commanding::Dispatcher.request(
    Commanding::Commands::Partitions::Pause.name,
    {
      consumer_group_id: consumer_group_id,
      topic: topic,
      partition_id: partition_id,
      # User provides this in seconds, we operate on ms in the system
      duration: params.int(:duration) * 1_000,
      prevent_override: params.bool(:prevent_override)
    },
    matchers: {
      consumer_group_id: consumer_group_id,
      topic: topic,
      partition_id: partition_id
    }
  )

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

#delete(consumer_group_id, topic, partition_id) ⇒ Object

Dispatches the partition resume command to all processes

Parameters:

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


107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
 
# File 'lib/karafka/web/pro/ui/controllers/consumers/partitions/pauses_controller.rb', line 107

def delete(consumer_group_id, topic, partition_id)
  new(consumer_group_id, topic, partition_id)

  # Broadcast to all processes with matchers to filter by consumer group,
  # topic, and partition
  Commanding::Dispatcher.request(
    Commanding::Commands::Partitions::Resume.name,
    {
      consumer_group_id: consumer_group_id,
      topic: topic,
      partition_id: partition_id,
      reset_attempts: params.bool(:reset_attempts)
    },
    matchers: {
      consumer_group_id: consumer_group_id,
      topic: topic,
      partition_id: partition_id
    }
  )

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

#edit(consumer_group_id, topic, partition_id) ⇒ Object

Displays the resume configuration form for a partition

Parameters:

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


96
97
98
99
100
 
# File 'lib/karafka/web/pro/ui/controllers/consumers/partitions/pauses_controller.rb', line 96

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

  render
end

#new(consumer_group_id, topic, partition_id) ⇒ Object

Displays the pause configuration form for a partition

Parameters:

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


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

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

  render
end