RemoteFetcherTierStatemachine logging

aiven · Jan 5, 2024 · c7eae32 · c7eae32
1 parent 815a323
commit c7eae32
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 3 deletions.
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java b/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java
@@ -127,7 +127,7 @@ public void run() {
                     maybeWaitForPartitionAssignments();
                 }
 
-                log.trace("Polling consumer to receive remote log metadata topic records");
+                log.debug("Polling consumer to receive remote log metadata topic records");
                 final ConsumerRecords<byte[], byte[]> consumerRecords = consumer.poll(Duration.ofMillis(pollTimeoutMs));
                 for (ConsumerRecord<byte[], byte[]> record : consumerRecords) {
                     processConsumerRecord(record);
@@ -153,11 +153,15 @@ public void run() {
 
     private void processConsumerRecord(ConsumerRecord<byte[], byte[]> record) {
         final RemoteLogMetadata remoteLogMetadata = serde.deserialize(record.value());
+        log.error("Received remote log metadata: {} from partition: {} with offset: {}",
+            remoteLogMetadata, record.partition(), record.offset());
         if (shouldProcess(remoteLogMetadata, record.offset())) {
+            log.error("Processing remote log metadata: {} from partition: {} with offset: {}",
+                remoteLogMetadata, record.partition(), record.offset());
             remotePartitionMetadataEventHandler.handleRemoteLogMetadata(remoteLogMetadata);
             readOffsetsByUserTopicPartition.put(remoteLogMetadata.topicIdPartition(), record.offset());
         } else {
-            log.trace("The event {} is skipped because it is either already processed or not assigned to this consumer",
+            log.debug("The event {} is skipped because it is either already processed or not assigned to this consumer",
                     remoteLogMetadata);
         }
         log.trace("Updating consumed offset: {} for partition {}", record.offset(), record.partition());
@@ -254,6 +258,7 @@ void maybeWaitForPartitionAssignments() throws InterruptedException {
                 if (!utp.isAssigned) {
                     // Note that there can be a race between `remove` and `add` partition assignment. Calling the
                     // `maybeLoadPartition` here again to be sure that the partition gets loaded on the handler.
+                    log.error("Loading partition {} from maybeWaitForPartitionAssignments", utp.topicIdPartition);
                     remotePartitionMetadataEventHandler.maybeLoadPartition(utp.topicIdPartition);
                     utp.isAssigned = true;
                 }
@@ -463,4 +468,4 @@ public String toString() {
                 '}';
         }
     }
-}
+}
diff --git a/...g/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java b/...g/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java
@@ -287,8 +287,11 @@ public void onPartitionLeadershipChanges(Set<TopicIdPartition> leaderPartitions,
             if (!initialized.get()) {
                 // If it is not yet initialized, then keep them as pending partitions and assign them
                 // when it is initialized successfully in initializeResources().
+                log.error("Pending assign partitions {}", pendingAssignPartitions);
                 this.pendingAssignPartitions.addAll(allPartitions);
+                log.error("added more partitions to pending {} .", allPartitions);
             } else {
+                log.error("Assigning partitions {} while topic-based RLMM is already initialized.", allPartitions);
                 assignPartitions(allPartitions);
             }
         } finally {
@@ -426,6 +429,8 @@ private void initializeResources() {
                     }
 
                     if (!pendingAssignPartitions.isEmpty()) {
+                        log.error("Pending assign partitions {} found while initializing topic-based RLMM resources.",
+                                pendingAssignPartitions);
                         assignPartitions(pendingAssignPartitions);
                         pendingAssignPartitions.clear();
                     }