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) {