Add partition offset methods for multiple OffsetSpecs

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
 {
@@ -34,7 +35,7 @@
         /// <inheritdoc cref="Admin.Metrics{ReturnExtendsOrg_Apache_Kafka_Common_Metric}"/>
         Map<MetricName, T> Metrics<T>() where T : Metric;
         /// <inheritdoc cref="Admin.Close()"/>
         void Close();
         /// <inheritdoc cref="Admin.Close(Duration)"/>
         void Close(Duration timeout);
         /// <inheritdoc cref="Admin.CreateTopics(Collection{NewTopic})"/>
@@ -231,10 +232,63 @@
         /// <returns>The unique cluster id</returns>
         string GetClusterId();
         /// <summary>
-        /// Returns a <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the last offset for each partition of the <paramref name="topicName"/>
+        /// Returns a <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the ForTimestamp offset for each partition of the <paramref name="topicName"/>
         /// </summary>
         /// <param name="topicName">The topic to be queried</param>
-        /// <returns>A <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the last offset for each partition</returns>
+        /// <param name="timestamp">The starting <see cref="DateTime"/></param>
+        /// <param name="delta">A value to be added to each offset retrieved</param>
+        /// <returns>A <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the ForTimestamp offset for each partition</returns>
+        System.Collections.Generic.IDictionary<int, long> ForTimestampPartitionOffsetForTopic(string topicName, DateTime timestamp, long delta = 0);
+        /// <summary>
+        /// Returns a <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the ForTimestamp offset for each partition of the <paramref name="topicName"/>
+        /// </summary>
+        /// <param name="topicName">The topic to be queried</param>
+        /// <param name="timestamp">The starting Unix time in milliseconds</param>
+        /// <param name="delta">A value to be added to each offset retrieved</param>
+        /// <returns>A <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the ForTimestamp offset for each partition</returns>
+        System.Collections.Generic.IDictionary<int, long> ForTimestampPartitionOffsetForTopic(string topicName, long timestamp, long delta = 0);
+        /// <summary>
+        /// Returns a <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the MaxTimestamp offset for each partition of the <paramref name="topicName"/>
+        /// </summary>
+        /// <param name="topicName">The topic to be queried</param>
+        /// <param name="delta">A value to be added to each offset retrieved</param>
+        /// <returns>A <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the MaxTimestamp offset for each partition</returns>
+        System.Collections.Generic.IDictionary<int, long> MaxTimestampPartitionOffsetForTopic(string topicName, long delta = 0);
+        /// <summary>
+        /// Returns a <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the Earliest offset for each partition of the <paramref name="topicName"/>
+        /// </summary>
+        /// <param name="topicName">The topic to be queried</param>
+        /// <param name="delta">A value to be added to each offset retrieved</param>
+        /// <returns>A <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the Earliest offset for each partition</returns>
+        System.Collections.Generic.IDictionary<int, long> EarliestPartitionOffsetForTopic(string topicName, long delta = 0);
+        /// <summary>
+        /// Returns a <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the EarliestLocal offset for each partition of the <paramref name="topicName"/>
+        /// </summary>
+        /// <param name="topicName">The topic to be queried</param>
+        /// <param name="delta">A value to be added to each offset retrieved</param>
+        /// <returns>A <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the EarliestLocal offset for each partition</returns>
+        System.Collections.Generic.IDictionary<int, long> EarliestLocalPartitionOffsetForTopic(string topicName, long delta = 0);
+        /// <summary>
+        /// Returns a <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the Latest offset for each partition of the <paramref name="topicName"/>
+        /// </summary>
+        /// <param name="topicName">The topic to be queried</param>
+        /// <param name="delta">A value to be added to each offset retrieved</param>
+        /// <returns>A <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the Latest offset for each partition</returns>
+        System.Collections.Generic.IDictionary<int, long> LatestPartitionOffsetForTopic(string topicName, long delta = -1);
+        /// <summary>
+        /// Returns a <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the LatestTiered offset for each partition of the <paramref name="topicName"/>
+        /// </summary>
+        /// <param name="topicName">The topic to be queried</param>
+        /// <param name="delta">A value to be added to each offset retrieved</param>
+        /// <returns>A <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the LatestTiered offset for each partition</returns>
+        System.Collections.Generic.IDictionary<int, long> LatestTieredPartitionOffsetForTopic(string topicName, long delta = -1);
+        /// <summary>
+        /// Returns a <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the Latest offset for each partition of the <paramref name="topicName"/>
+        /// </summary>
+        /// <param name="topicName">The topic to be queried</param>
+        /// <param name="delta">A value to be added to each offset retrieved</param>
+        /// <returns>A <see cref="System.Collections.Generic.IDictionary{TKey, TValue}"/> containing the Latest offset for each partition</returns>
+        [Obsolete("Use LatestTieredPartitionOffsetForTopic", true)]
         System.Collections.Generic.IDictionary<int, long> LastPartitionOffsetForTopic(string topicName);
     }
 
@@ -258,7 +312,62 @@
         }
 
         /// <inheritdoc/>
+        public System.Collections.Generic.IDictionary<int, long> ForTimestampPartitionOffsetForTopic(string topicName, DateTime timestamp, long delta = 0)
+        {
+            return ForTimestampPartitionOffsetForTopic(topicName, new System.DateTimeOffset(timestamp).ToUnixTimeMilliseconds());
+        }
+
+        /// <inheritdoc/>
+        public System.Collections.Generic.IDictionary<int, long> ForTimestampPartitionOffsetForTopic(string topicName, long timestamp, long delta = 0)
+        {
+            using var offsetSpec = OffsetSpec.ForTimestamp(timestamp);
+            return PartitionOffsetForTopic(topicName, offsetSpec, delta);
+        }
+
+        /// <inheritdoc/>
+        public System.Collections.Generic.IDictionary<int, long> MaxTimestampPartitionOffsetForTopic(string topicName, long delta = 0)
+        {
+            using var offsetSpec = OffsetSpec.MaxTimestamp();
+            return PartitionOffsetForTopic(topicName, offsetSpec, delta);
+        }
+
+        /// <inheritdoc/>
+        public System.Collections.Generic.IDictionary<int, long> EarliestPartitionOffsetForTopic(string topicName, long delta = 0)
+        {
+            using var offsetSpec = OffsetSpec.Earliest();
+            return PartitionOffsetForTopic(topicName, offsetSpec, delta);
+        }
+
+        /// <inheritdoc/>
+        public System.Collections.Generic.IDictionary<int, long> EarliestLocalPartitionOffsetForTopic(string topicName, long delta = 0)
+        {
+            using var offsetSpec = OffsetSpec.EarliestLocal();
+            return PartitionOffsetForTopic(topicName, offsetSpec, delta);
+        }
+
+        /// <inheritdoc/>
+        public System.Collections.Generic.IDictionary<int, long> LatestPartitionOffsetForTopic(string topicName, long delta = -1)
+        {
+            using var offsetSpec = OffsetSpec.Latest();
+            return PartitionOffsetForTopic(topicName, offsetSpec, delta);
+        }
+
+        /// <inheritdoc/>
+        public System.Collections.Generic.IDictionary<int, long> LatestTieredPartitionOffsetForTopic(string topicName, long delta = -1)
+        {
+            using var offsetSpec = OffsetSpec.LatestTiered();
+            return PartitionOffsetForTopic(topicName, offsetSpec, delta);
+        }
+
+        /// <inheritdoc/>
+        [Obsolete("Use LatestTieredPartitionOffsetForTopic", true)]
         public System.Collections.Generic.IDictionary<int, long> LastPartitionOffsetForTopic(string topicName)
+        {
+            using var offsetSpec = OffsetSpec.Latest();
+            return PartitionOffsetForTopic(topicName, offsetSpec, -1);
+        }
+
+        System.Collections.Generic.IDictionary<int, long> PartitionOffsetForTopic(string topicName, OffsetSpec offsetSpec, long delta)
         {
             System.Collections.Generic.Dictionary<int, long> dictionary = new();
             try
@@ -285,7 +394,6 @@
                                 {
                                     var partitionIndex = partition.Partition();
                                     using TopicPartition topicPartition = new(topicName, partitionIndex);
-                                    using var offsetSpec = OffsetSpec.Latest();
                                     hashMap.Put(topicPartition, offsetSpec).DisposeIfDisposable();
                                 }
                             }
@@ -301,7 +409,7 @@
                                 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;

tests/net/General/KNetTest/Program.cs

@@ -313,7 +313,7 @@ static IDictionary<int, long> 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)
             {