chore(trino): Backport Kafka offset handling to 477 (#1373)

* chore(trino): Backport Kafka offset handling to 477

Backport trinodb/trino#26789 to Trino 477

* changelog

Author: sbernauer

CHANGELOG.md

@@ -19,6 +19,7 @@ All notable changes to this project will be documented in this file.
 - testing-tools: refactoring: Split image into multiple images, remove unnecessary components and switch to UBI as base image ([#1354]).
 - hive: fixed 4.0.1 shaded hive-metastore-opa-authorizer jar by relocating dependencies ([#1356]).
 - testing-tools: fix: add kubectl and openssl ([#1367]).
+- trino: Backport Kafka offset handling to 477 ([#1373]).
 - ubi: Bumped ubi9 and ubi10 hashes ([#1386]).
 
 ### Removed
@@ -35,6 +36,7 @@ All notable changes to this project will be documented in this file.
 [#1358]: https://github.com/stackabletech/docker-images/pull/1358
 [#1366]: https://github.com/stackabletech/docker-images/pull/1366
 [#1367]: https://github.com/stackabletech/docker-images/pull/1367
+[#1373]: https://github.com/stackabletech/docker-images/pull/1373
 [#1386]: https://github.com/stackabletech/docker-images/pull/1386
 
 ## [25.11.0] - 2025-11-07

trino/trino/stackable/patches/477/0003-Fix-findOffsetsForTimestampGreaterOrEqual.patch

@@ -0,0 +1,26 @@
+From a46ead926108b561ca6fa6b193dd38db5a8cb125 Mon Sep 17 00:00:00 2001
+From: Sebastian Bernauer <sebastian.bernauer@stackable.tech>
+Date: Wed, 7 Jan 2026 13:30:17 +0100
+Subject: Fix findOffsetsForTimestampGreaterOrEqual
+
+Back-port of https://github.com/trinodb/trino/pull/26789
+
+Co-authored-by: Mateusz "Serafin" Gajewski <github@wendigo.pl>
+---
+ .../main/java/io/trino/plugin/kafka/KafkaFilterManager.java    | 3 +--
+ 1 file changed, 1 insertion(+), 2 deletions(-)
+
+diff --git a/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/KafkaFilterManager.java b/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/KafkaFilterManager.java
+index 5c853d9011..25d3c83301 100644
+--- a/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/KafkaFilterManager.java
++++ b/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/KafkaFilterManager.java
+@@ -182,8 +182,7 @@ public class KafkaFilterManager
+     {
+         Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestamps = kafkaConsumer.offsetsForTimes(timestamps);
+         return topicPartitionOffsetAndTimestamps.entrySet().stream()
+-                .collect(toMap(Map.Entry::getKey, entry -> Optional.of(entry.getValue())
+-                        .map(OffsetAndTimestamp::offset)));
++                .collect(toMap(Map.Entry::getKey, entry -> Optional.ofNullable(entry.getValue()).map(OffsetAndTimestamp::offset)));
+     }
+ 
+     private static Map<TopicPartition, Long> overridePartitionBeginOffsets(Map<TopicPartition, Long> partitionBeginOffsets,