Fix list offsets for times failure when ledgers are removed by a rollover operation (#1854)

### Motivation
When lac ledger is expired or
removed，KafkaRequestHandler.fetchOffsetByTimestamp call failed and
`NonRecoverableLedgerException` will be thrown.
- When lac ledger is expired or removed，the first legerId in
managedLedger is greater than the lac ledgerId , then the
`asyncFindNewestMatching.startPosition` is lac'sLedgerId:0.
- `OpFindNewestEntry.find` will read `startPosition`'s entry, then
managedLedger will thrown `NonRecoverableLedgerException`

### Modifications

- Handle the NonRecoverableLedgerException in findEntryFailed callback.
In this case, return the LEO as the offset because we can treat the
topic as empty.

(cherry picked from commit c0ece19)

Author: zhouxy0809

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java

@@ -1396,9 +1396,16 @@ public void findEntryComplete(Position position, Object ctx) {
             @Override
             public void findEntryFailed(ManagedLedgerException exception,
                                         Optional<Position> position, Object ctx) {
-                log.warn("Unable to find position for topic {} time {}. Exception:",
-                        topic, timestamp, exception);
-                partitionData.complete(Pair.of(Errors.UNKNOWN_SERVER_ERROR, null));
+                if (exception instanceof ManagedLedgerException.NonRecoverableLedgerException) {
+                    // The position doesn't exist, it usually happens when the rollover of managed ledger leads to
+                    // the deletion of all expired ledgers. In this case, there's only one empty ledger in the managed
+                    // ledger. So here we complete it with the latest offset.
+                    partitionData.complete(Pair.of(Errors.NONE, MessageMetadataUtils.getLogEndOffset(managedLedger)));
+                } else {
+                    log.warn("Unable to find position for topic {} time {}. Exception:",
+                            topic, timestamp, exception);
+                    partitionData.complete(Pair.of(Errors.UNKNOWN_SERVER_ERROR, null));
+                }
             }
         });
     }

tests/src/test/java/io/streamnative/pulsar/handlers/kop/MultiLedgerTest.java

@@ -37,6 +37,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
@@ -238,6 +239,18 @@ public void testListOffsetForEmptyRolloverLedger() throws Exception {
             fail(e.getMessage());
         }
 
+        // Verify listing offsets for timestamp return a correct offset
+        try {
+            final Map<TopicPartition, OffsetAndTimestamp> partitionToTimestamp =
+                    consumer.offsetsForTimes(Collections.singletonMap(new TopicPartition(topic, 0), 0L),
+                            Duration.ofSeconds(2));
+            assertTrue(partitionToTimestamp.containsKey(topicPartition));
+            assertEquals(partitionToTimestamp.get(topicPartition).offset(), numMessages);
+        } catch (Exception e) {
+            log.error("Failed to get offsets for times: {}", e.getMessage());
+            fail(e.getMessage());
+        }
+
         // Verify consumer can start consuming from the correct position
         consumer.subscribe(Collections.singleton(topic));
         try (final KafkaProducer<String, String> producer = new KafkaProducer<>(newKafkaProducerProperties());) {