root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-17406: Move ClientIdAndBroker to server-common module (#16967) 

Reviewers: Mickael Maison <mickael.maison@gmail.com>, David Arthur <mumrah@gmail.com>
This commit is contained in:
Dmitry Werner 2024-09-04 14:04:40 +05:00 committed by GitHub
parent 5fd7ce2ace
commit 005155ba5c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 20 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
core/src/main/scala/kafka/server/AbstractFetcherThread.scala
View File

@ -18,7 +18,6 @@
package kafka.server
import com.yammer.metrics.core.Meter
import kafka.common.ClientIdAndBroker
import kafka.server.AbstractFetcherThread.{ReplicaFetch, ResultWithPartitions}
import kafka.utils.CoreUtils.inLock
import kafka.utils.Implicits._
@ -32,7 +31,7 @@ import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.record.{FileRecords, MemoryRecords, Records}
import org.apache.kafka.common.requests.OffsetsForLeaderEpochResponse.{UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET}
import org.apache.kafka.common.requests._
import org.apache.kafka.common.{InvalidRecordException, TopicPartition, Uuid}
import org.apache.kafka.common.{ClientIdAndBroker, InvalidRecordException, TopicPartition, Uuid}
import org.apache.kafka.server.common.OffsetAndEpoch
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.util.ShutdownableThread
@ -72,7 +71,7 @@ abstract class AbstractFetcherThread(name: String,
protected val partitionMapLock = new ReentrantLock
private val partitionMapCond = partitionMapLock.newCondition()
private val metricId = ClientIdAndBroker(clientId, leader.brokerEndPoint().host, leader.brokerEndPoint().port)
private val metricId = new ClientIdAndBroker(clientId, leader.brokerEndPoint().host, leader.brokerEndPoint().port)
val fetcherStats = new FetcherStats(metricId)
val fetcherLagStats = new FetcherLagStats(metricId)

28
core/src/main/scala/kafka/common/ClientIdAndBroker.scala → server-common/src/main/java/org/apache/kafka/common/ClientIdAndBroker.java
View File

@ -1,14 +1,12 @@
package kafka.common
/**
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
@ -16,15 +14,25 @@ package kafka.common
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.common;
/**
* Convenience case class since (clientId, brokerInfo) pairs are used to create
* SyncProducer Request Stats and SimpleConsumer Request and Response Stats.
*/
public class ClientIdAndBroker {
public final String clientId;
public final String brokerHost;
public final int brokerPort;
trait ClientIdBroker {
}
public ClientIdAndBroker(String clientId, String brokerHost, int brokerPort) {
this.clientId = clientId;
this.brokerHost = brokerHost;
this.brokerPort = brokerPort;
}
case class ClientIdAndBroker(clientId: String, brokerHost: String, brokerPort: Int) extends ClientIdBroker {
override def toString: String = "%s-%s-%d".format(clientId, brokerHost, brokerPort)
@Override
public String toString() {
return String.format("%s-%s-%d", clientId, brokerHost, brokerPort);
}
}