From 3ab73c50003fd68c2c9e00d9a7f294e311805cad Mon Sep 17 00:00:00 2001
From: Carol <947752894@qq.com>
Date: Thu, 5 Dec 2019 10:32:15 +0800
Subject: [PATCH] add new method to get offset from kafka
---
.idea/workspace.xml | 226 +++++++++++++------
src/main/java/site/cnkj/utils/KafkaUtil.java | 33 +++
2 files changed, 192 insertions(+), 67 deletions(-)
diff --git a/.idea/workspace.xml b/.idea/workspace.xml
index 4792b69..d1d8a71 100644
--- a/.idea/workspace.xml
+++ b/.idea/workspace.xml
@@ -3,6 +3,7 @@
+
@@ -17,7 +18,7 @@
-
+
@@ -26,10 +27,10 @@
-
+
-
+
@@ -38,6 +39,50 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
@@ -52,12 +97,12 @@
+
-
@@ -77,6 +122,9 @@
listRemove
public long rightPush
MongoClient
+ group
+ get
+ BOOTSTRAP_SERVERS_CONFIG
@@ -97,6 +145,7 @@
+
@@ -153,9 +202,6 @@
-
-
-
@@ -211,6 +257,9 @@
+
+
+
@@ -252,7 +301,7 @@
-
+
@@ -264,6 +313,17 @@
+
+
+
+
+
+
+
+
+
+
+
@@ -305,8 +365,13 @@
+
+
+
@@ -324,6 +389,8 @@
+
+
1574320188766
@@ -416,14 +483,22 @@
1574819212177
-
+
+ 1574835930667
+
+
+
+ 1574835930667
+
+
-
+
+
@@ -438,21 +513,21 @@
+
-
-
+
+
-
+
-
@@ -541,59 +616,11 @@
-
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
@@ -869,7 +896,7 @@
-
+
@@ -895,16 +922,16 @@
+
-
-
+
@@ -912,6 +939,71 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/src/main/java/site/cnkj/utils/KafkaUtil.java b/src/main/java/site/cnkj/utils/KafkaUtil.java
index a20d9ab..08e2454 100644
--- a/src/main/java/site/cnkj/utils/KafkaUtil.java
+++ b/src/main/java/site/cnkj/utils/KafkaUtil.java
@@ -11,6 +11,12 @@ import java.util.*;
*/
public class KafkaUtil {
+ private static final String BOOTSTRAP_SERVERS = "bootstrap.servers";
+ private static final String GROUP_ID = "group.id";
+ private static final String KEY_DESERIALIZER = "key.deserializer";
+ private static final String VALUE_DESERIALIZER = "value.deserializer";
+ private static final String STRING_DESERIALIZER = "org.apache.kafka.common.serialization.StringDeserializer";
+
/**
* 获取当前topic下的全部分区的偏移量信息
@@ -72,4 +78,31 @@ public class KafkaUtil {
}
}
+ /**
+ * 获取指定集群上指定groupId的topic的每个分区的当前偏移量
+ *
+ * @param bootstrapServers 消费地址
+ * @param groupId id
+ * @param topics topic集合
+ * @return {
+ * topic:
+ * {
+ * partitionInfo:offset
+ * }
+ * }
+ */
+ public static Map> getConsumerOffset(List bootstrapServers, String groupId, List topics){
+ try {
+ Map properties = new HashMap<>();
+ properties.put(BOOTSTRAP_SERVERS, bootstrapServers);
+ properties.put(GROUP_ID, groupId);
+ properties.put(KEY_DESERIALIZER, STRING_DESERIALIZER);
+ properties.put(VALUE_DESERIALIZER, STRING_DESERIALIZER);
+ return getConsumerTopicPartitionsOffset(properties, new HashSet(topics));
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ return null;
+ }
+
}