made committable data members into readonly structs

src/Akka.Streams.Kafka/Messages/CommittedMarker.cs

@@ -24,23 +24,35 @@ internal interface ICommittedMarker
     /// <summary>
     /// Used by <see cref="TransactionalMessageBuilder{K,V}"/>
     /// </summary>
-    internal sealed record PartitionOffsetCommittedMarker : GroupTopicPartitionOffset
+    internal sealed record PartitionOffsetCommittedMarker
     {
         /// <summary>
         /// Committed marker
         /// </summary>
         public ICommittedMarker CommittedMarker { get; }
 
         public PartitionOffsetCommittedMarker(string groupId, string topic, int partition, Offset offset, ICommittedMarker committedMarker) 
-            : base(groupId, topic, partition, offset)
+            : this(new GroupTopicPartitionOffset(groupId, topic, partition, offset), committedMarker)
         {
             CommittedMarker = committedMarker;
         }
 
-        public PartitionOffsetCommittedMarker(GroupTopicPartition groupTopicPartition, Offset offset, ICommittedMarker committedMarker) 
-            : base(groupTopicPartition, offset)
+        public PartitionOffsetCommittedMarker(GroupTopicPartitionOffset groupTopicPartition, ICommittedMarker committedMarker)
         {
             CommittedMarker = committedMarker;
+            GroupTopicPartitionOffset = groupTopicPartition;
         }
+
+        public GroupTopicPartitionOffset GroupTopicPartitionOffset { get; }
+
+        public string GroupId => GroupTopicPartitionOffset.GroupId;
+
+        public string Topic => GroupTopicPartitionOffset.Topic;
+
+        public int Partition => GroupTopicPartitionOffset.Partition;
+
+        public Offset Offset => GroupTopicPartitionOffset.Offset;
+
+        public GroupTopicPartition GroupTopicPartition => GroupTopicPartitionOffset.GroupTopicPartition;
     }
 }

src/Akka.Streams.Kafka/Messages/GroupTopicPartitionOffset.cs

@@ -6,57 +6,46 @@ namespace Akka.Streams.Kafka.Messages
     /// <summary>
     /// Offset position for a groupId, topic, partition.
     /// </summary>
-    public record GroupTopicPartitionOffset
+    public readonly record struct GroupTopicPartitionOffset(GroupTopicPartition GroupTopicPartition, Offset Offset)
     {
-        /// <summary>
-        /// GroupTopicPartitionOffset
-        /// </summary>
         public GroupTopicPartitionOffset(string groupId, string topic, int partition, Offset offset)
-        {
-            GroupId = groupId;
-            Topic = topic;
-            Partition = partition;
-            Offset = offset;
-        }
-
-        /// <summary>
-        /// GroupTopicPartitionOffset
-        /// </summary>
-        public GroupTopicPartitionOffset(GroupTopicPartition groupTopicPartition, Offset offset)
-            : this(groupTopicPartition.GroupId, groupTopicPartition.Topic, groupTopicPartition.Partition, offset)
+            : this(new GroupTopicPartition(groupId, topic, partition), offset)
         {
         }
 
         /// <summary>
         /// Consumer's group Id
         /// </summary>
-        public string GroupId { get; }
+        public string GroupId => GroupTopicPartition.GroupId;
+
         /// <summary>
         /// Topic
         /// </summary>
-        public string Topic { get; }
+        public string Topic => GroupTopicPartition.Topic;
+
         /// <summary>
         /// Partition
         /// </summary>
-        public int Partition { get; }
+        public int Partition => GroupTopicPartition.Partition;
+
         /// <summary>
         /// Kafka partition offset value
         /// </summary>
-        public Offset Offset { get; }
-        
+        public Offset Offset { get; } = Offset;
+
         /// <summary>
         /// Group topic partition info
         /// </summary>
-        public GroupTopicPartition GroupTopicPartition => new(GroupId, Topic, Partition);
+        public GroupTopicPartition GroupTopicPartition { get; } = GroupTopicPartition;
     }
 
     /// <summary>
     /// Group, topic and partition info
     /// </summary>
-    public sealed record GroupTopicPartition(string GroupId, string Topic, int Partition)
+    public readonly record struct GroupTopicPartition(string GroupId, string Topic, int Partition)
     {
         public TopicPartition TopicPartition { get; } = new(Topic, Partition);
     }
 
-    public sealed record OffsetAndMetadata(Offset Offset, string Metadata);
+    public readonly record struct OffsetAndMetadata(Offset Offset, string Metadata);
 }

src/Akka.Streams.Kafka/Stages/Consumers/MessageBuilders.cs

@@ -172,7 +172,7 @@ public TransactionalMessage<K, V> CreateMessage(ConsumeResult<K, V> record)
                 record.Offset, 
                 _transactionalMessageBuilderStage.CommittedMarker);
 
-            return new TransactionalMessage<K, V>(record, offset);
+            return new TransactionalMessage<K, V>(record, offset.GroupTopicPartitionOffset);
         }
     }
 }