Fix wrong offset resolution

CHANGELOG.md

@@ -6,6 +6,16 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
 <!-- insertion marker -->
+## Unreleased
+
+<small>[Compare with latest](https://github.com/tradewelltech/beavers/compare/v0.9.1...HEAD)</small>
+
+### Fixed
+
+- Fix wrong offset resolution ([16f3dad](https://github.com/tradewelltech/beavers/commit/16f3dad52b819d2981310e15f4fb2b0db36d2086) by aandres3).
+
+<!-- insertion marker -->
+
 ## [v0.9.1](https://github.com/tradewelltech/beavers/releases/tag/v0.9.1) - 2024-09-20
 
 <small>[Compare with v0.9.0](https://github.com/tradewelltech/beavers/compare/v0.9.0...v0.9.1)</small>

beavers/kafka.py

@@ -705,8 +705,9 @@ def _resolve_offset_for_time(
     )
     return {
         confluent_kafka.TopicPartition(topic=tp.topic, partition=tp.partition): (
-            tp.offset if tp.offset > 0 else (watermarks[tp][1] - 1),
-            watermarks[tp][1] - 1,
+            (tp.offset, watermarks[tp][1] - 1)
+            if tp.offset >= 0
+            else (watermarks[tp][1], watermarks[tp][1])
         )
         for tp in offset_for_time
     }

scripts/README.md

@@ -21,6 +21,7 @@ The "timestamp" of the output messages should be in order across topics when rep
 ```shell
 docker exec -it simple_kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server=localhost:9092 --create --topic=left --partitions=1 --replication-factor=1 
 docker exec -it simple_kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server=localhost:9092 --create --topic=right --partitions=1 --replication-factor=1 
+docker exec -it simple_kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server=localhost:9092 --create --topic=both --partitions=1 --replication-factor=1 
 ```
 
 ### Run the Beavers job
@@ -43,7 +44,7 @@ docker exec -it simple_kafka /opt/kafka/bin/kafka-console-consumer.sh \
   --bootstrap-server=localhost:9092 \
   --topic=both \
   --property print.key=true \
-  --from-beginning    
+  --from-beginning
 ```
 
 ## `perpective_test_bench.py`

tests/test_kafka.py

@@ -1407,7 +1407,7 @@ def test_resolve_offset_for_time_minus_one():
     consumer._offsets_for_time[(tp1, millis)] = TopicPartition("topic-1", 0, -1)
     assert _resolve_offset_for_time(
         timestamp, consumer, {tp1: (40, 80)}, timeout=1.0
-    ) == {tp1: (79, 79)}
+    ) == {tp1: (80, 80)}
 
     consumer._offsets_for_time[(tp1, millis)] = TopicPartition("topic-1", 0, 70)
     assert _resolve_offset_for_time(