KAFKA-17476 Delete kafka.common.OffsetAndMetadata (apache#17553)

Reviewers: Chia-Ping Tsai <[email protected]>

Author: mimaison

core/src/main/scala/kafka/common/OffsetAndMetadata.scala

@@ -18,7 +18,6 @@ package kafka.coordinator.group
 
 import java.util.{OptionalInt, Properties}
 import java.util.concurrent.atomic.AtomicBoolean
-import kafka.common.OffsetAndMetadata
 import kafka.server._
 import kafka.utils.Logging
 import org.apache.kafka.common.{TopicIdPartition, TopicPartition}
@@ -32,7 +31,7 @@ import org.apache.kafka.common.protocol.{ApiKeys, Errors}
 import org.apache.kafka.common.record.RecordBatch
 import org.apache.kafka.common.requests._
 import org.apache.kafka.common.utils.Time
-import org.apache.kafka.coordinator.group.{Group, OffsetConfig}
+import org.apache.kafka.coordinator.group.{Group, OffsetAndMetadata, OffsetConfig}
 import org.apache.kafka.server.common.RequestLocal
 import org.apache.kafka.server.record.BrokerCompressionType
 import org.apache.kafka.storage.internals.log.VerificationGuard

core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala

@@ -16,7 +16,6 @@
  */
 package kafka.coordinator.group
 
-import kafka.common.OffsetAndMetadata
 import kafka.server.{KafkaConfig, ReplicaManager}
 import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
 import org.apache.kafka.common.message.{ConsumerGroupDescribeResponseData, ConsumerGroupHeartbeatRequestData, ConsumerGroupHeartbeatResponseData, DeleteGroupsResponseData, DescribeGroupsResponseData, HeartbeatRequestData, HeartbeatResponseData, JoinGroupRequestData, JoinGroupResponseData, LeaveGroupRequestData, LeaveGroupResponseData, ListGroupsRequestData, ListGroupsResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteRequestData, OffsetDeleteResponseData, OffsetFetchRequestData, OffsetFetchResponseData, ShareGroupDescribeResponseData, ShareGroupHeartbeatRequestData, ShareGroupHeartbeatResponseData, SyncGroupRequestData, SyncGroupResponseData, TxnOffsetCommitRequestData, TxnOffsetCommitResponseData}
@@ -26,13 +25,14 @@ import org.apache.kafka.common.record.RecordBatch
 import org.apache.kafka.common.requests.{OffsetCommitRequest, RequestContext, TransactionResult}
 import org.apache.kafka.common.utils.{BufferSupplier, Time}
 import org.apache.kafka.coordinator.group
+import org.apache.kafka.coordinator.group.OffsetAndMetadata
 import org.apache.kafka.image.{MetadataDelta, MetadataImage}
 import org.apache.kafka.server.common.RequestLocal
 import org.apache.kafka.server.util.FutureUtils
 
 import java.time.Duration
 import java.util
-import java.util.{Optional, OptionalInt, Properties}
+import java.util.{Optional, OptionalInt, OptionalLong, Properties}
 import java.util.concurrent.CompletableFuture
 import java.util.function.IntSupplier
 import scala.collection.{immutable, mutable}
@@ -504,20 +504,23 @@ private[group] class GroupCoordinatorAdapter(
     expireTimestamp: Option[Long]
   ): OffsetAndMetadata = {
     new OffsetAndMetadata(
-      offset = offset,
-      leaderEpoch = leaderEpoch match {
-        case RecordBatch.NO_PARTITION_LEADER_EPOCH => Optional.empty[Integer]
-        case committedLeaderEpoch => Optional.of[Integer](committedLeaderEpoch)
+      offset,
+      leaderEpoch match {
+        case RecordBatch.NO_PARTITION_LEADER_EPOCH => OptionalInt.empty
+        case committedLeaderEpoch => OptionalInt.of(committedLeaderEpoch)
       },
-      metadata = metadata match {
-        case null => OffsetAndMetadata.NoMetadata
+      metadata match {
+        case null => OffsetAndMetadata.NO_METADATA
         case metadata => metadata
       },
-      commitTimestamp = commitTimestamp match {
+      commitTimestamp match {
         case OffsetCommitRequest.DEFAULT_TIMESTAMP => currentTimeMs
         case customTimestamp => customTimestamp
       },
-      expireTimestamp = expireTimestamp
+      expireTimestamp match {
+        case Some(timestamp) => OptionalLong.of(timestamp)
+        case None => OptionalLong.empty()
+      }
     )
   }
 

core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala

@@ -19,16 +19,14 @@ package kafka.coordinator.group
 import java.nio.ByteBuffer
 import java.util.UUID
 import java.util.concurrent.locks.ReentrantLock
-
-import kafka.common.OffsetAndMetadata
 import kafka.utils.{CoreUtils, Logging, nonthreadsafe}
 import org.apache.kafka.clients.consumer.internals.ConsumerProtocol
 import org.apache.kafka.common.{TopicIdPartition, TopicPartition}
 import org.apache.kafka.common.message.JoinGroupResponseData.JoinGroupResponseMember
 import org.apache.kafka.common.protocol.Errors
 import org.apache.kafka.common.protocol.types.SchemaException
 import org.apache.kafka.common.utils.Time
-import org.apache.kafka.coordinator.group.Group
+import org.apache.kafka.coordinator.group.{Group, OffsetAndMetadata}
 
 import scala.collection.{Seq, immutable, mutable}
 import scala.jdk.CollectionConverters._
@@ -761,13 +759,12 @@ private[group] class GroupMetadata(val groupId: String, initialState: GroupState
         case (topicPartition, commitRecordMetadataAndOffset) =>
           !subscribedTopics.contains(topicPartition.topic()) &&
           !pendingOffsetCommits.contains(topicPartition) && {
-            commitRecordMetadataAndOffset.offsetAndMetadata.expireTimestamp match {
-              case None =>
-                // current version with no per partition retention
-                currentTimestamp - baseTimestamp(commitRecordMetadataAndOffset) >= offsetRetentionMs
-              case Some(expireTimestamp) =>
-                // older versions with explicit expire_timestamp field => old expiration semantics is used
-                currentTimestamp >= expireTimestamp
+            if (commitRecordMetadataAndOffset.offsetAndMetadata.expireTimestampMs.isEmpty) {
+              // current version with no per partition retention
+              currentTimestamp - baseTimestamp(commitRecordMetadataAndOffset) >= offsetRetentionMs
+            } else {
+              // older versions with explicit expire_timestamp field => old expiration semantics is used
+              currentTimestamp >= commitRecordMetadataAndOffset.offsetAndMetadata.expireTimestampMs.getAsLong
             }
           }
       }.map {
@@ -785,7 +782,7 @@ private[group] class GroupMetadata(val groupId: String, initialState: GroupState
         //   since the last commit timestamp, expire the offset
         getExpiredOffsets(
           commitRecordMetadataAndOffset => currentStateTimestamp
-            .getOrElse(commitRecordMetadataAndOffset.offsetAndMetadata.commitTimestamp)
+            .getOrElse(commitRecordMetadataAndOffset.offsetAndMetadata.commitTimestampMs)
         )
 
       case Some(ConsumerProtocol.PROTOCOL_TYPE) if subscribedTopics.isDefined && is(Stable) =>
@@ -794,14 +791,14 @@ private[group] class GroupMetadata(val groupId: String, initialState: GroupState
         //   the last commit timestamp, expire the offset. offset with pending offset commit are not
         //   expired
         getExpiredOffsets(
-          _.offsetAndMetadata.commitTimestamp,
+          _.offsetAndMetadata.commitTimestampMs,
           subscribedTopics.get
         )
 
       case None =>
         // protocolType is None => standalone (simple) consumer, that uses Kafka for offset storage only
         // expire offsets with no pending offset commit that retention period has passed since their last commit
-        getExpiredOffsets(_.offsetAndMetadata.commitTimestamp)
+        getExpiredOffsets(_.offsetAndMetadata.commitTimestampMs)
 
       case _ =>
         Map()

core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala

@@ -18,13 +18,12 @@
 package kafka.coordinator.group
 
 import java.nio.ByteBuffer
-import java.util.{Optional, OptionalInt}
+import java.util.{Optional, OptionalInt, OptionalLong}
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.locks.ReentrantLock
 import java.util.concurrent.ConcurrentHashMap
 import java.util.function.Supplier
 import com.yammer.metrics.core.Gauge
-import kafka.common.OffsetAndMetadata
 import kafka.coordinator.group.GroupMetadataManager.maybeConvertOffsetCommitError
 import kafka.server.ReplicaManager
 import kafka.utils.CoreUtils.inLock
@@ -40,7 +39,7 @@ import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
 import org.apache.kafka.common.requests.{OffsetCommitRequest, OffsetFetchResponse}
 import org.apache.kafka.common.utils.{Time, Utils}
 import org.apache.kafka.common.{TopicIdPartition, TopicPartition}
-import org.apache.kafka.coordinator.group.OffsetConfig
+import org.apache.kafka.coordinator.group.{OffsetAndMetadata, OffsetConfig}
 import org.apache.kafka.coordinator.group.generated.{GroupMetadataValue, OffsetCommitKey, OffsetCommitValue, GroupMetadataKey => GroupMetadataKeyData}
 import org.apache.kafka.server.common.{MetadataVersion, RequestLocal}
 import org.apache.kafka.server.common.MetadataVersion.{IBP_0_10_1_IV0, IBP_2_1_IV0, IBP_2_1_IV1, IBP_2_3_IV0}
@@ -523,8 +522,12 @@ class GroupMetadataManager(brokerId: Int,
                   new PartitionData(OffsetFetchResponse.INVALID_OFFSET,
                     Optional.empty(), "", Errors.NONE)
                 case Some(offsetAndMetadata) =>
-                  new PartitionData(offsetAndMetadata.offset,
-                    offsetAndMetadata.leaderEpoch, offsetAndMetadata.metadata, Errors.NONE)
+                  val leaderEpoch: Optional[Integer] = if (offsetAndMetadata.leaderEpoch.isPresent) {
+                    Optional.of(offsetAndMetadata.leaderEpoch.getAsInt)
+                  } else {
+                    Optional.empty()
+                  }
+                  new PartitionData(offsetAndMetadata.committedOffset, leaderEpoch, offsetAndMetadata.metadata, Errors.NONE)
               }
               topicPartition -> partitionData
             }
@@ -1087,18 +1090,18 @@ object GroupMetadataManager {
   def offsetCommitValue(offsetAndMetadata: OffsetAndMetadata,
                         metadataVersion: MetadataVersion): Array[Byte] = {
     val version =
-      if (metadataVersion.isLessThan(IBP_2_1_IV0) || offsetAndMetadata.expireTimestamp.nonEmpty) 1.toShort
+      if (metadataVersion.isLessThan(IBP_2_1_IV0) || offsetAndMetadata.expireTimestampMs.isPresent) 1.toShort
       else if (metadataVersion.isLessThan(IBP_2_1_IV1)) 2.toShort
       // Serialize with the highest supported non-flexible version
       // until a tagged field is introduced or the version is bumped.
       else 3.toShort
     MessageUtil.toVersionPrefixedBytes(version, new OffsetCommitValue()
-      .setOffset(offsetAndMetadata.offset)
+      .setOffset(offsetAndMetadata.committedOffset)
       .setMetadata(offsetAndMetadata.metadata)
-      .setCommitTimestamp(offsetAndMetadata.commitTimestamp)
+      .setCommitTimestamp(offsetAndMetadata.commitTimestampMs)
       .setLeaderEpoch(offsetAndMetadata.leaderEpoch.orElse(RecordBatch.NO_PARTITION_LEADER_EPOCH))
       // version 1 has a non empty expireTimestamp field
-      .setExpireTimestamp(offsetAndMetadata.expireTimestamp.getOrElse(OffsetCommitRequest.DEFAULT_TIMESTAMP))
+      .setExpireTimestamp(offsetAndMetadata.expireTimestampMs.orElse(OffsetCommitRequest.DEFAULT_TIMESTAMP))
     )
   }
 
@@ -1179,12 +1182,12 @@ object GroupMetadataManager {
       val version = buffer.getShort
       if (version >= OffsetCommitValue.LOWEST_SUPPORTED_VERSION && version <= OffsetCommitValue.HIGHEST_SUPPORTED_VERSION) {
         val value = new OffsetCommitValue(new ByteBufferAccessor(buffer), version)
-        OffsetAndMetadata(
-          offset = value.offset,
-          leaderEpoch = if (value.leaderEpoch == RecordBatch.NO_PARTITION_LEADER_EPOCH) Optional.empty() else Optional.of(value.leaderEpoch),
-          metadata = value.metadata,
-          commitTimestamp = value.commitTimestamp,
-          expireTimestamp = if (value.expireTimestamp == OffsetCommitRequest.DEFAULT_TIMESTAMP) None else Some(value.expireTimestamp))
+        new OffsetAndMetadata(
+          value.offset,
+          if (value.leaderEpoch == RecordBatch.NO_PARTITION_LEADER_EPOCH) OptionalInt.empty() else OptionalInt.of(value.leaderEpoch),
+          value.metadata,
+          value.commitTimestamp,
+          if (value.expireTimestamp == OffsetCommitRequest.DEFAULT_TIMESTAMP) OptionalLong.empty() else OptionalLong.of(value.expireTimestamp))
       } else throw new IllegalStateException(s"Unknown offset message version: $version")
     }
   }

core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala

@@ -16,7 +16,6 @@
  */
 package kafka.coordinator.group
 
-import kafka.common.OffsetAndMetadata
 import kafka.coordinator.group.GroupCoordinatorConcurrencyTest.{JoinGroupCallback, SyncGroupCallback}
 import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
 import org.apache.kafka.common.errors.{InvalidGroupIdException, UnsupportedVersionException}
@@ -31,6 +30,7 @@ import org.apache.kafka.common.requests.{OffsetFetchResponse, RequestContext, Re
 import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
 import org.apache.kafka.common.utils.{BufferSupplier, Time}
 import org.apache.kafka.common.utils.annotation.ApiKeyVersionsSource
+import org.apache.kafka.coordinator.group.OffsetAndMetadata
 import org.apache.kafka.server.common.RequestLocal
 import org.apache.kafka.server.util.MockTime
 import org.apache.kafka.test.TestUtils.assertFutureThrows
@@ -41,7 +41,7 @@ import org.mockito.{ArgumentCaptor, ArgumentMatchers}
 import org.mockito.Mockito.{mock, verify, when}
 
 import java.net.InetAddress
-import java.util.Optional
+import java.util.{Optional, OptionalInt, OptionalLong}
 import scala.jdk.CollectionConverters._
 
 class GroupCoordinatorAdapterTest {
@@ -696,11 +696,11 @@ class GroupCoordinatorAdapterTest {
       ArgumentMatchers.eq(data.generationIdOrMemberEpoch),
       ArgumentMatchers.eq(Map(
         new TopicIdPartition(Uuid.ZERO_UUID, 0 , "foo") -> new OffsetAndMetadata(
-          offset = 100,
-          leaderEpoch = Optional.of[Integer](1),
-          metadata = "",
-          commitTimestamp = now,
-          expireTimestamp = Some(now + 1000L)
+          100,
+          OptionalInt.of(1),
+          "",
+          now,
+          OptionalLong.of(now + 1000L)
         )
       )),
       capturedCallback.capture(),
@@ -769,11 +769,11 @@ class GroupCoordinatorAdapterTest {
       ArgumentMatchers.eq(data.generationId),
       ArgumentMatchers.eq(Map(
         new TopicIdPartition(Uuid.ZERO_UUID, 0 , "foo") -> new OffsetAndMetadata(
-          offset = 100,
-          leaderEpoch = Optional.of[Integer](1),
-          metadata = "",
-          commitTimestamp = now,
-          expireTimestamp = None
+          100,
+          OptionalInt.of(1),
+          "",
+          now,
+          OptionalLong.empty()
         )
       )),
       capturedCallback.capture(),