Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions core/src/main/scala/kafka/server/SharedServer.scala
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ class SharedServer(
@volatile var brokerMetrics: BrokerServerMetrics = _
@volatile var controllerServerMetrics: ControllerMetadataMetrics = _
@volatile var loader: MetadataLoader = _
val snapshotsDiabledReason = new AtomicReference[String](null)
val snapshotsDisabledReason = new AtomicReference[String](null)
@volatile var snapshotEmitter: SnapshotEmitter = _
@volatile var snapshotGenerator: SnapshotGenerator = _

Expand Down Expand Up @@ -166,7 +166,7 @@ class SharedServer(
action = () => SharedServer.this.synchronized {
Option(brokerMetrics).foreach(_.metadataLoadErrorCount.getAndIncrement())
Option(controllerServerMetrics).foreach(_.incrementMetadataErrorCount())
snapshotsDiabledReason.compareAndSet(null, "metadata loading fault")
snapshotsDisabledReason.compareAndSet(null, "metadata loading fault")
})

/**
Expand All @@ -177,7 +177,7 @@ class SharedServer(
fatal = true,
action = () => SharedServer.this.synchronized {
Option(controllerServerMetrics).foreach(_.incrementMetadataErrorCount())
snapshotsDiabledReason.compareAndSet(null, "controller startup fault")
snapshotsDisabledReason.compareAndSet(null, "controller startup fault")
})

/**
Expand All @@ -189,7 +189,7 @@ class SharedServer(
action = () => SharedServer.this.synchronized {
Option(brokerMetrics).foreach(_.metadataApplyErrorCount.getAndIncrement())
Option(controllerServerMetrics).foreach(_.incrementMetadataErrorCount())
snapshotsDiabledReason.compareAndSet(null, "initial broker metadata loading fault")
snapshotsDisabledReason.compareAndSet(null, "initial broker metadata loading fault")
})

/**
Expand All @@ -200,7 +200,7 @@ class SharedServer(
fatal = true,
action = () => SharedServer.this.synchronized {
Option(controllerServerMetrics).foreach(_.incrementMetadataErrorCount())
snapshotsDiabledReason.compareAndSet(null, "quorum controller fault")
snapshotsDisabledReason.compareAndSet(null, "quorum controller fault")
})

/**
Expand Down Expand Up @@ -269,7 +269,7 @@ class SharedServer(
setFaultHandler(metadataPublishingFaultHandler).
setMaxBytesSinceLastSnapshot(sharedServerConfig.metadataSnapshotMaxNewRecordBytes).
setMaxTimeSinceLastSnapshotNs(TimeUnit.MILLISECONDS.toNanos(sharedServerConfig.metadataSnapshotMaxIntervalMs)).
setDisabledReason(snapshotsDiabledReason).
setDisabledReason(snapshotsDisabledReason).
setThreadNamePrefix(s"kafka-${sharedServerConfig.nodeId}-").
build()
_raftManager.register(loader)
Expand Down