From 2b4abf248b61c477a6636c1f26b1dafae46009a5 Mon Sep 17 00:00:00 2001 From: masesdevelopers <94312179+masesdevelopers@users.noreply.github.com> Date: Fri, 7 Aug 2026 23:17:17 +0200 Subject: [PATCH 1/4] Add partition offset methods for multiple OffsetSpecs Introduce multiple partition-offset helpers: ForTimestampPartitionOffsetForTopic (DateTime and unix-ms), MaxTimestamp, Earliest, EarliestLocal, EarliestPendingUpload, Latest and LatestTiered. Refactor PartitionOffsetForTopic to accept an OffsetSpec and a delta adjustment (used to normalize latest offsets), and update offset computation accordingly. Also add System using. This replaces the prior LastPartitionOffsetForTopic with more explicit OffsetSpec-based methods and centralizes disposal/OffsetSpec handling. --- .../Org/Apache/Kafka/Clients/Admin/Admin.cs | 118 +++++++++++++++++- 1 file changed, 112 insertions(+), 6 deletions(-) diff --git a/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs b/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs index 0f55c3a165..95db67ef51 100644 --- a/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs +++ b/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs @@ -26,6 +26,7 @@ using Org.Apache.Kafka.Common.Acl; using Org.Apache.Kafka.Common.Config; using Org.Apache.Kafka.Common.Quota; +using System; namespace Org.Apache.Kafka.Clients.Admin { @@ -231,11 +232,63 @@ public partial interface IAdmin /// The unique cluster id string GetClusterId(); /// - /// Returns a containing the last offset for each partition of the + /// Returns a containing the ForTimestamp offset for each partition of the /// /// The topic to be queried - /// A containing the last offset for each partition - System.Collections.Generic.IDictionary LastPartitionOffsetForTopic(string topicName); + /// The starting + /// A value to be added to each offset retrieved + /// A containing the ForTimestamp offset for each partition + System.Collections.Generic.IDictionary ForTimestampPartitionOffsetForTopic(string topicName, DateTime timestamp, long delta = 0); + /// + /// Returns a containing the ForTimestamp offset for each partition of the + /// + /// The topic to be queried + /// The starting Unix time in milliseconds + /// A value to be added to each offset retrieved + /// A containing the ForTimestamp offset for each partition + System.Collections.Generic.IDictionary ForTimestampPartitionOffsetForTopic(string topicName, long timestamp, long delta = 0); + /// + /// Returns a containing the MaxTimestamp offset for each partition of the + /// + /// The topic to be queried + /// A value to be added to each offset retrieved + /// A containing the MaxTimestamp offset for each partition + System.Collections.Generic.IDictionary MaxTimestampPartitionOffsetForTopic(string topicName, long delta = 0); + /// + /// Returns a containing the Earliest offset for each partition of the + /// + /// The topic to be queried + /// A value to be added to each offset retrieved + /// A containing the Earliest offset for each partition + System.Collections.Generic.IDictionary EarliestPartitionOffsetForTopic(string topicName, long delta = 0); + /// + /// Returns a containing the EarliestLocal offset for each partition of the + /// + /// The topic to be queried + /// A value to be added to each offset retrieved + /// A containing the EarliestLocal offset for each partition + System.Collections.Generic.IDictionary EarliestLocalPartitionOffsetForTopic(string topicName, long delta = 0); + /// + /// Returns a containing the EarliestPendingUpload offset for each partition of the + /// + /// The topic to be queried + /// A value to be added to each offset retrieved + /// A containing the EarliestPendingUpload offset for each partition + System.Collections.Generic.IDictionary EarliestPendingUploadPartitionOffsetForTopic(string topicName, long delta = 0); + /// + /// Returns a containing the Latest offset for each partition of the + /// + /// The topic to be queried + /// A value to be added to each offset retrieved + /// A containing the Latest offset for each partition + System.Collections.Generic.IDictionary LatestPartitionOffsetForTopic(string topicName, long delta = -1); + /// + /// Returns a containing the LatestTiered offset for each partition of the + /// + /// The topic to be queried + /// A value to be added to each offset retrieved + /// A containing the LatestTiered offset for each partition + System.Collections.Generic.IDictionary LatestTieredPartitionOffsetForTopic(string topicName, long delta = -1); } public partial class Admin @@ -258,7 +311,61 @@ public string GetClusterId() } /// - public System.Collections.Generic.IDictionary LastPartitionOffsetForTopic(string topicName) + public System.Collections.Generic.IDictionary ForTimestampPartitionOffsetForTopic(string topicName, DateTime timestamp, long delta = 0) + { + return ForTimestampPartitionOffsetForTopic(topicName, new System.DateTimeOffset(timestamp).ToUnixTimeMilliseconds()); + } + + /// + public System.Collections.Generic.IDictionary ForTimestampPartitionOffsetForTopic(string topicName, long timestamp, long delta = 0) + { + using var offsetSpec = OffsetSpec.ForTimestamp(timestamp); + return PartitionOffsetForTopic(topicName, offsetSpec, delta); + } + + /// + public System.Collections.Generic.IDictionary MaxTimestampPartitionOffsetForTopic(string topicName, long delta = 0) + { + using var offsetSpec = OffsetSpec.MaxTimestamp(); + return PartitionOffsetForTopic(topicName, offsetSpec, delta); + } + + /// + public System.Collections.Generic.IDictionary EarliestPartitionOffsetForTopic(string topicName, long delta = 0) + { + using var offsetSpec = OffsetSpec.Earliest(); + return PartitionOffsetForTopic(topicName, offsetSpec, delta); + } + + /// + public System.Collections.Generic.IDictionary EarliestLocalPartitionOffsetForTopic(string topicName, long delta = 0) + { + using var offsetSpec = OffsetSpec.EarliestLocal(); + return PartitionOffsetForTopic(topicName, offsetSpec, delta); + } + + /// + public System.Collections.Generic.IDictionary EarliestPendingUploadPartitionOffsetForTopic(string topicName, long delta = 0) + { + using var offsetSpec = OffsetSpec.EarliestPendingUpload(); + return PartitionOffsetForTopic(topicName, offsetSpec, delta); + } + + /// + public System.Collections.Generic.IDictionary LatestPartitionOffsetForTopic(string topicName, long delta = -1) + { + using var offsetSpec = OffsetSpec.Latest(); + return PartitionOffsetForTopic(topicName, offsetSpec, delta); + } + + /// + public System.Collections.Generic.IDictionary LatestTieredPartitionOffsetForTopic(string topicName, long delta = -1) + { + using var offsetSpec = OffsetSpec.LatestTiered(); + return PartitionOffsetForTopic(topicName, offsetSpec, delta); + } + + System.Collections.Generic.IDictionary PartitionOffsetForTopic(string topicName, OffsetSpec offsetSpec, long delta) { System.Collections.Generic.Dictionary dictionary = new(); try @@ -285,7 +392,6 @@ public System.Collections.Generic.IDictionary LastPartitionOffsetForT { var partitionIndex = partition.Partition(); using TopicPartition topicPartition = new(topicName, partitionIndex); - using var offsetSpec = OffsetSpec.Latest(); hashMap.Put(topicPartition, offsetSpec).DisposeIfDisposable(); } } @@ -301,7 +407,7 @@ public System.Collections.Generic.IDictionary LastPartitionOffsetForT using var offsetResultItemTopic = offsetResultItemKey.Topic(); if (offsetResultItemTopic.Equals(jTopic)) { - dictionary.Add(offsetResultItemKey.Partition(), offsetResultItemValue.Offset() - 1); // since latest means the latest used offset (a record in kafka) + 1, here we remove 1 to be in sync with received offset from kafka + dictionary.Add(offsetResultItemKey.Partition(), offsetResultItemValue.Offset() + delta); // since latest means the latest used offset (a record in kafka) + 1, here we remove 1 to be in sync with received offset from kafka } } break; From 8314d8812373aeffada3bfc0f98cfdad5591f272 Mon Sep 17 00:00:00 2001 From: masesdevelopers <94312179+masesdevelopers@users.noreply.github.com> Date: Fri, 7 Aug 2026 23:28:51 +0200 Subject: [PATCH 2/4] Update test and reintroduce obsolete old function --- .../Org/Apache/Kafka/Clients/Admin/Admin.cs | 16 ++++++++++++++++ tests/net/General/KNetTest/Program.cs | 2 +- 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs b/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs index 95db67ef51..33ad9d8626 100644 --- a/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs +++ b/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs @@ -289,6 +289,14 @@ public partial interface IAdmin /// A value to be added to each offset retrieved /// A containing the LatestTiered offset for each partition System.Collections.Generic.IDictionary LatestTieredPartitionOffsetForTopic(string topicName, long delta = -1); + /// + /// Returns a containing the Latest offset for each partition of the + /// + /// The topic to be queried + /// A value to be added to each offset retrieved + /// A containing the Latest offset for each partition + [Obsolete("Use LatestTieredPartitionOffsetForTopic", true)] + System.Collections.Generic.IDictionary LastPartitionOffsetForTopic(string topicName); } public partial class Admin @@ -365,6 +373,14 @@ public System.Collections.Generic.IDictionary LatestTieredPartitionOf return PartitionOffsetForTopic(topicName, offsetSpec, delta); } + /// + [Obsolete("Use LatestTieredPartitionOffsetForTopic", true)] + System.Collections.Generic.IDictionary LastPartitionOffsetForTopic(string topicName) + { + using var offsetSpec = OffsetSpec.Latest(); + return PartitionOffsetForTopic(topicName, offsetSpec, -1); + } + System.Collections.Generic.IDictionary PartitionOffsetForTopic(string topicName, OffsetSpec offsetSpec, long delta) { System.Collections.Generic.Dictionary dictionary = new(); diff --git a/tests/net/General/KNetTest/Program.cs b/tests/net/General/KNetTest/Program.cs index d7f82e405a..0cb5887ed7 100644 --- a/tests/net/General/KNetTest/Program.cs +++ b/tests/net/General/KNetTest/Program.cs @@ -313,7 +313,7 @@ static IDictionary LastOffsetOfTopic(string topicName) Console.WriteLine($"LastOffsetOfTopic for {topicName} using an AdminClient based on {props}"); using IAdmin admin = KafkaAdminClient.Create(props); - return admin.LastPartitionOffsetForTopic(topicName); + return admin.LatestPartitionOffsetForTopic(topicName); } catch (Java.Util.Concurrent.ExecutionException ex) { From abf1dfa7b806330d1c51bceeed6917fb54626d69 Mon Sep 17 00:00:00 2001 From: MASES Public Developers Team <94312179+masesdevelopers@users.noreply.github.com> Date: Fri, 7 Aug 2026 21:31:12 +0000 Subject: [PATCH 3/4] Removed not applicable OffsetSpec --- .../Org/Apache/Kafka/Clients/Admin/Admin.cs | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs b/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs index 33ad9d8626..59a39daa7c 100644 --- a/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs +++ b/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs @@ -269,13 +269,6 @@ public partial interface IAdmin /// A containing the EarliestLocal offset for each partition System.Collections.Generic.IDictionary EarliestLocalPartitionOffsetForTopic(string topicName, long delta = 0); /// - /// Returns a containing the EarliestPendingUpload offset for each partition of the - /// - /// The topic to be queried - /// A value to be added to each offset retrieved - /// A containing the EarliestPendingUpload offset for each partition - System.Collections.Generic.IDictionary EarliestPendingUploadPartitionOffsetForTopic(string topicName, long delta = 0); - /// /// Returns a containing the Latest offset for each partition of the /// /// The topic to be queried @@ -352,13 +345,6 @@ public System.Collections.Generic.IDictionary EarliestLocalPartitionO return PartitionOffsetForTopic(topicName, offsetSpec, delta); } - /// - public System.Collections.Generic.IDictionary EarliestPendingUploadPartitionOffsetForTopic(string topicName, long delta = 0) - { - using var offsetSpec = OffsetSpec.EarliestPendingUpload(); - return PartitionOffsetForTopic(topicName, offsetSpec, delta); - } - /// public System.Collections.Generic.IDictionary LatestPartitionOffsetForTopic(string topicName, long delta = -1) { From de5244e6258e9a66c60f464b59597ea08a7b9b2e Mon Sep 17 00:00:00 2001 From: masesdevelopers <94312179+masesdevelopers@users.noreply.github.com> Date: Fri, 7 Aug 2026 23:32:14 +0200 Subject: [PATCH 4/4] Fix implementation --- src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs b/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs index 59a39daa7c..56f6198e15 100644 --- a/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs +++ b/src/net/KNet/Developed/Org/Apache/Kafka/Clients/Admin/Admin.cs @@ -361,7 +361,7 @@ public System.Collections.Generic.IDictionary LatestTieredPartitionOf /// [Obsolete("Use LatestTieredPartitionOffsetForTopic", true)] - System.Collections.Generic.IDictionary LastPartitionOffsetForTopic(string topicName) + public System.Collections.Generic.IDictionary LastPartitionOffsetForTopic(string topicName) { using var offsetSpec = OffsetSpec.Latest(); return PartitionOffsetForTopic(topicName, offsetSpec, -1);