jiazhizhong/kafka-console-ui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix按时间查询消息bug,加长页面请求超时

Browse Source
This commit is contained in:
许晓东
2021-12-13 19:05:57 +08:00
parent 57a41e087f
commit b9548d1640
6 changed files with 37 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/main/scala/kafka/console/MessageConsole.scala
View File

@@ -55,25 +55,26 @@ class MessageConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConf
} else {
val records = consumer.poll(Duration.ofMillis(timeoutMs))
for ((tp, endOff) <- endOffTable) {
if (!terminate) {
val recordList = records.records(tp)
if (recordList.isEmpty) {
arrive.remove(tp)
} else {
val first = recordList.get(0)
if (first.offset() >= endOff) {
arrive.remove(tp)
} else {
res.addAll(recordList)
if (recordList.get(recordList.size() - 1).offset() >= endOff) {
if (records.isEmpty) {
terminate = true
} else {
for ((tp, endOff) <- endOffTable) {
if (!terminate) {
val recordList = records.records(tp)
if (!recordList.isEmpty) {
val first = recordList.get(0)
if (first.offset() >= endOff) {
arrive.remove(tp)
} else {
res.addAll(recordList)
if (recordList.get(recordList.size() - 1).offset() >= endOff) {
arrive.remove(tp)
}
}
}
}
if (arrive.isEmpty) {
terminate = true;
if (arrive.isEmpty) {
terminate = true;
}
}
}
}