Index: core/src/main/scala/kafka/log/LogCleaner.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- core/src/main/scala/kafka/log/LogCleaner.scala	(revision 77a89fcf8d7fa01831683327727f11f4bed97319)
+++ core/src/main/scala/kafka/log/LogCleaner.scala	(date 1578667996310)
@@ -942,11 +942,16 @@
             stats.indexMessagesRead(batch.countOrNull)
           } else {
             for (record <- batch.asScala) {
-              if (record.hasKey && record.offset >= startOffset) {
-                if (map.size < maxDesiredMapSize)
-                  map.put(record.key, record.offset)
-                else
-                  return true
+              if (record.hasKey) {
+                if (record.offset >= startOffset) {
+                  if (map.size < maxDesiredMapSize)
+                    map.put(record.key, record.offset)
+                  else
+                    return true
+                } else {
+                  // KAFKA-8764 - LogCleanerManager endless loop while compacting/cleaning segments
+                  map.updateLatestOffset(record.offset + 1)
+                }
               }
               stats.indexMessagesRead(1)
             }