查看topic分区消息偏移

src/main/scala/kafka/console/KafkaConsole.scala

@@ -6,7 +6,9 @@ import com.xuxd.kafka.console.config.KafkaConfig
 import kafka.zk.{AdminZkClient, KafkaZkClient}
 import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.admin.{Admin, AdminClientConfig}
+import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
 import org.apache.kafka.common.config.SaslConfigs
+import org.apache.kafka.common.serialization.ByteArrayDeserializer
 import org.apache.kafka.common.utils.Time
 
 /**
@@ -37,6 +39,22 @@ class KafkaConsole(config: KafkaConfig) {
         }
     }
 
+    protected def withConsumerAndCatchError(f: KafkaConsumer[Array[Byte], Array[Byte]] => Any, eh: Exception => Any,
+        extra: Properties = new Properties()): Any = {
+        val props = getProps()
+        props.putAll(extra)
+        props.put(ConsumerConfig.CLIENT_ID_CONFIG, String.valueOf(System.currentTimeMillis()))
+        val consumer = new KafkaConsumer(props, new ByteArrayDeserializer, new ByteArrayDeserializer)
+        try {
+            f(consumer)
+        } catch {
+            case er: Exception => eh(er)
+        }
+        finally {
+            consumer.close()
+        }
+    }
+
     protected def withZKClient(f: AdminZkClient => Any): Any = {
         val zkClient = KafkaZkClient(config.getZookeeperAddr, false, 30000, 30000, Int.MaxValue, Time.SYSTEM)
         val adminZkClient = new AdminZkClient(zkClient)
@@ -48,6 +66,10 @@ class KafkaConsole(config: KafkaConfig) {
     }
 
     private def createAdminClient(): Admin = {
+        Admin.create(getProps())
+    }
+
+    private def getProps(): Properties = {
         val props: Properties = new Properties();
         props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServer)
         props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, config.getRequestTimeoutMs())
@@ -56,7 +78,6 @@ class KafkaConsole(config: KafkaConfig) {
             props.put(SaslConfigs.SASL_MECHANISM, config.getSaslMechanism())
             props.put(SaslConfigs.SASL_JAAS_CONFIG, config.getSaslJaasConfig())
         }
-
-        Admin.create(props)
+        props
     }
 }

src/main/scala/kafka/console/TopicConsole.scala

@@ -6,6 +6,7 @@ import java.util.{Collections, List, Set}
 
 import com.xuxd.kafka.console.config.KafkaConfig
 import org.apache.kafka.clients.admin.{DeleteTopicsOptions, ListTopicsOptions, TopicDescription}
+import org.apache.kafka.common.TopicPartition
 
 import scala.jdk.CollectionConverters.{CollectionHasAsScala, SetHasAsJava}
 
@@ -72,4 +73,24 @@ class TopicConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConfig
                 (false, e.getMessage)
             }).asInstanceOf[(Boolean, String)]
     }
+
+    /**
+     * get topic begin offset and end offset.
+     *
+     * @param topic      topic name.
+     * @param partitions topic partition info list.
+     * @return partition ->  begin offset and end offset.
+     */
+    def getTopicOffset(topic: String,
+        partitions: List[TopicPartition]): (util.Map[TopicPartition, Long], util.Map[TopicPartition, Long]) = {
+
+        withConsumerAndCatchError(consumer => {
+            val beginOffsets = consumer.beginningOffsets(partitions)
+            val endOffsets = consumer.endOffsets(partitions)
+            (beginOffsets, endOffsets)
+        }, e => {
+            log.error("getTopicOffset error, topic: " + topic, e)
+            (Collections.emptyMap(), Collections.emptyMap())
+        }).asInstanceOf[(util.Map[TopicPartition, Long], util.Map[TopicPartition, Long])]
+    }
 }