RemoteFetcherTierStatemachine logging

aiven · Jan 5, 2024 · df946bb · df946bb
1 parent 020b330
commit df946bb
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 2 deletions.
diff --git a/core/src/main/java/kafka/server/ReplicaFetcherTierStateMachine.java b/core/src/main/java/kafka/server/ReplicaFetcherTierStateMachine.java
@@ -97,6 +97,7 @@ public PartitionFetchState start(TopicPartition topicPartition,
 
         log.error("Starting the tier state machine for partition: {} with currentFetchState: {} and fetchPartitionData: {}",
                 topicPartition, currentFetchState, fetchPartitionData);
+        log.error("the leader is: {}", leader);
         OffsetAndEpoch epochAndLeaderLocalStartOffset = leader.fetchEarliestLocalOffset(topicPartition, currentFetchState.currentLeaderEpoch());
         log.error("Fetched the earliest local offset for partition: {} with epochAndLeaderLocalStartOffset: {}", topicPartition, epochAndLeaderLocalStartOffset);
         int epoch = epochAndLeaderLocalStartOffset.leaderEpoch();

diff --git a/core/src/main/scala/kafka/server/LocalLeaderEndPoint.scala b/core/src/main/scala/kafka/server/LocalLeaderEndPoint.scala
@@ -132,7 +132,9 @@ class LocalLeaderEndPoint(sourceBroker: BrokerEndPoint,
   override def fetchEarliestLocalOffset(topicPartition: TopicPartition, currentLeaderEpoch: Int): OffsetAndEpoch = {
     val partition = replicaManager.getPartitionOrException(topicPartition)
     val localLogStartOffset = partition.localLogOrException.localLogStartOffset()
+    error(s"Fetching earliest local offset for $topicPartition returned $localLogStartOffset and partition $partition")
     val epoch = partition.localLogOrException.leaderEpochCache.get.epochForOffset(localLogStartOffset)
+    error(s"Got epoch $epoch from ${partition.localLogOrException.leaderEpochCache}")
     new OffsetAndEpoch(localLogStartOffset, epoch.orElse(0))
   }
 

diff --git a/...ava/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataStore.java b/...ava/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataStore.java
@@ -151,7 +151,8 @@ private FileBasedRemoteLogMetadataCache getRemoteLogMetadataCache(TopicIdPartiti
         if (remoteLogMetadataCache == null) {
             throw new RemoteResourceNotFoundException("No resource found for partition: " + topicIdPartition);
         }
-
+        log.error("remoteLogMetadataCache.isInitialized(): {}", remoteLogMetadataCache.isInitialized());
+        log.error("remoteLogMetadataCache: {}", remoteLogMetadataCache);
         if (!remoteLogMetadataCache.isInitialized()) {
             // Throwing a retriable ReplicaNotAvailableException here for clients retry. We can introduce a new more
             // appropriate exception with a KIP in the future.
@@ -196,7 +197,7 @@ public void maybeLoadPartition(TopicIdPartition partition) {
     @Override
     public void markInitialized(TopicIdPartition partition) {
         idToRemoteLogMetadataCache.get(partition).markInitialized();
-        log.trace("Remote log components are initialized for user-partition: {}", partition);
+        log.error("Remote log components are initialized for user-partition: {}", partition);
     }
 
     @Override