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..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
@@ -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,10 +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
+ /// 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 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);
+ ///
+ /// 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);
}
@@ -258,7 +312,62 @@ public string GetClusterId()
}
///
+ 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 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);
+ }
+
+ ///
+ [Obsolete("Use LatestTieredPartitionOffsetForTopic", true)]
public 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();
try
@@ -285,7 +394,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 +409,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;
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)
{