Class: Kazoo::Partition

Inherits:
Object
  • Object
show all
Defined in:
lib/kazoo/partition.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(topic, id, replicas: nil) ⇒ Partition

Returns a new instance of Partition.



5
6
7
8
 
# File 'lib/kazoo/partition.rb', line 5

def initialize(topic, id, replicas: nil)
  @topic, @id, @replicas = topic, id, replicas
  @mutex = Mutex.new
end

Instance Attribute Details

#id ⇒ Object (readonly)

Returns the value of attribute id.



3
4
5
 
# File 'lib/kazoo/partition.rb', line 3

def id
  @id
end

#replicas ⇒ Object (readonly)

Returns the value of attribute replicas.



3
4
5
 
# File 'lib/kazoo/partition.rb', line 3

def replicas
  @replicas
end

#topic ⇒ Object (readonly)

Returns the value of attribute topic.



3
4
5
 
# File 'lib/kazoo/partition.rb', line 3

def topic
  @topic
end

Instance Method Details

#cluster ⇒ Object 



10
11
12
 
# File 'lib/kazoo/partition.rb', line 10

def cluster
  topic.cluster
end

#eql?(other) ⇒ Boolean Also known as: ==

Returns:

  • (Boolean) 


57
58
59
 
# File 'lib/kazoo/partition.rb', line 57

def eql?(other)
  other.kind_of?(Kazoo::Partition) && topic == other.topic && id == other.id
end

#hash ⇒ Object 



63
64
65
 
# File 'lib/kazoo/partition.rb', line 63

def hash
  [topic, id].hash
end

#inspect ⇒ Object 



49
50
51
 
# File 'lib/kazoo/partition.rb', line 49

def inspect
  "#<Kazoo::Partition #{topic.name}/#{id}>"
end

#isr ⇒ Object 



25
26
27
28
29
30
 
# File 'lib/kazoo/partition.rb', line 25

def isr
  @mutex.synchronize do
    refresh_state if @isr.nil?
    @isr
  end
end

#key ⇒ Object 



53
54
55
 
# File 'lib/kazoo/partition.rb', line 53

def key
  "#{topic.name}/#{id}"
end

#leader ⇒ Object 



18
19
20
21
22
23
 
# File 'lib/kazoo/partition.rb', line 18

def leader
  @mutex.synchronize do
    refresh_state if @leader.nil?
    @leader
  end
end

#replication_factor ⇒ Object 



14
15
16
 
# File 'lib/kazoo/partition.rb', line 14

def replication_factor
  replicas.length
end

#under_replicated? ⇒ Boolean

Returns:

  • (Boolean) 


32
33
34
 
# File 'lib/kazoo/partition.rb', line 32

def under_replicated?
  isr.length < replication_factor
end

#valid? ⇒ Boolean

Returns:

  • (Boolean) 


43
44
45
46
47
 
# File 'lib/kazoo/partition.rb', line 43

def valid?
  validate
rescue Kazoo::ValidationError
  false
end

#validate ⇒ Object

Raises:



36
37
38
39
40
41
 
# File 'lib/kazoo/partition.rb', line 36

def validate
  raise Kazoo::ValidationError, "No replicas defined for #{topic.name}/#{id}" if replicas.length == 0
  raise Kazoo::ValidationError, "The replicas of #{topic.name}/#{id} should be assigned to different brokers" if replicas.length > replicas.uniq.length

  true
end

#wait_for_leader ⇒ Object 



67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/kazoo/partition.rb', line 67

def wait_for_leader
  current_leader = nil
  while current_leader.nil?
    current_leader = begin
      leader
    rescue Kazoo::Error
      nil
    end

    sleep(0.1) if current_leader.nil?
  end
end