Skip to content

fix(core): update HW with LEO after creating log #497

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Nov 30, 2023
Merged

fix(core): update HW with LEO after creating log #497

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
edb95a0
refactor: change enable method name
mooc9988 Nov 30, 2023
0d15d89
fix(core): update HW with LEO
mooc9988 Nov 30, 2023
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
6 changes: 5 additions & 1 deletion core/src/main/scala/kafka/cluster/Partition.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -415,12 +415,16 @@ class Partition(val topicPartition: TopicPartition,
try {
val log = logManager.getOrCreateLog(topicPartition, isNew, isFutureReplica, topicId, newLeaderEpoch)
maybeLog = Some(log)
updateHighWatermark(log)
// AutoMQ for Kafka inject start
log match {
case elasticUnifiedLog: ElasticUnifiedLog =>
elasticUnifiedLog.confirmOffsetChangeListener = Some(() => handleLeaderConfirmOffsetMove())
// just update LEO to HW since we only have one replica
val initialHighWatermark = log.logEndOffset
log.updateHighWatermark(log.logEndOffset)
info(s"Log loaded for partition $topicPartition with initial high watermark $initialHighWatermark")
case _ =>
updateHighWatermark(log)
}
// AutoMQ for Kafka inject end
log
Expand Down
8 changes: 4 additions & 4 deletions core/src/main/scala/kafka/log/streamaspect/ElasticLogManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,7 @@ class ElasticLogManager(val client: Client) extends Logging {
object ElasticLogManager {
var INSTANCE: Option[ElasticLogManager] = None
var NAMESPACE = ""
private var turnedOn = false
private var isEnabled = false

def init(config: KafkaConfig, clusterId: String, broker: BrokerServer = null): Boolean = {
if (!config.elasticStreamEnabled) {
Expand Down Expand Up @@ -153,11 +153,11 @@ object ElasticLogManager {
true
}

def mark(shouldEnable: Boolean): Unit = {
turnedOn = shouldEnable
def enable(shouldEnable: Boolean): Unit = {
isEnabled = shouldEnable
}

def enabled(): Boolean = turnedOn
def enabled(): Boolean = isEnabled

def removeLog(topicPartition: TopicPartition): Unit = {
INSTANCE.get.removeLog(topicPartition)
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/server/BrokerServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,7 +204,7 @@ class BrokerServer(

// AutoMQ for Kafka inject start
// ElasticLogManager should be marked before LogManager is created.
ElasticLogManager.mark(config.elasticStreamEnabled)
ElasticLogManager.enable(config.elasticStreamEnabled)
// AutoMQ for Kafka inject end

// Create log manager, but don't start it because we need to delay any potential unclean shutdown log recovery
Expand Down