Add the missing PartitionGroupMetadataFetcher change

Author: Xin Gao

pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java

@@ -99,6 +99,7 @@ private Boolean fetchSingleStream()
   private Boolean fetchMultipleStreams()
       throws Exception {
     int numStreams = _streamConfigs.size();
+    int permanentTopicIndex = 0;
     for (int i = 0; i < numStreams; i++) {
       StreamConfig streamConfig = _streamConfigs.get(i);
       String topicName = streamConfig.getTopicName();
@@ -107,21 +108,32 @@ private Boolean fetchMultipleStreams()
               + topicName;
       StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
       int index = i;
+      int finalPermanentTopicIndex = permanentTopicIndex;
+      // For permanent topics, we use the index of the stream config to get the partition group consumption status.
+      // For ephemeral backfill topics, we use the topic name to filter the partition group consumption status.
       List<PartitionGroupConsumptionStatus> topicPartitionGroupConsumptionStatusList =
           _partitionGroupConsumptionStatusList.stream()
-              .filter(partitionGroupConsumptionStatus -> IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
-                  partitionGroupConsumptionStatus.getPartitionGroupId()) == index)
+              .filter(partitionGroupConsumptionStatus -> _streamConfigs.get(index).isEphemeralBackfillTopic()
+                  ? _streamConfigs.get(index).getTopicName().equals(partitionGroupConsumptionStatus.getTopicName())
+                  : IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
+                      partitionGroupConsumptionStatus.getPartitionGroupId()) == finalPermanentTopicIndex)
               .collect(Collectors.toList());
       try (StreamMetadataProvider streamMetadataProvider = streamConsumerFactory.createStreamMetadataProvider(
           StreamConsumerFactory.getUniqueClientId(clientId))) {
+        // Similarly, for ephemeral backfill topics, we create the partition group metadata with the topic name.
         _newPartitionGroupMetadataList.addAll(
             streamMetadataProvider.computePartitionGroupMetadata(clientId,
-                    streamConfig, topicPartitionGroupConsumptionStatusList, /*maxWaitTimeMs=*/15000,
+                    _streamConfigs.get(i), topicPartitionGroupConsumptionStatusList, /*maxWaitTimeMs=*/15000,
                     _forceGetOffsetFromStream)
                 .stream()
                 .map(metadata -> new PartitionGroupMetadata(
-                    IngestionConfigUtils.getPinotPartitionIdFromStreamPartitionId(metadata.getPartitionGroupId(),
-                        index), metadata.getStartOffset()))
+                    _streamConfigs.get(finalPermanentTopicIndex).isEphemeralBackfillTopic() ? _streamConfigs.get(index)
+                        .getTopicName() : "",
+                    _streamConfigs.get(finalPermanentTopicIndex).isEphemeralBackfillTopic()
+                        ? metadata.getPartitionGroupId()
+                        : IngestionConfigUtils.getPinotPartitionIdFromStreamPartitionId(metadata.getPartitionGroupId(),
+                            finalPermanentTopicIndex),
+                    metadata.getStartOffset()))
                 .collect(Collectors.toList()));
         if (_exception != null) {
           // We had at least one failure, but succeeded now. Log an info
@@ -136,6 +148,7 @@ private Boolean fetchMultipleStreams()
         _exception = e;
         throw e;
       }
+      permanentTopicIndex += _streamConfigs.get(i).isEphemeralBackfillTopic() ? 0 : 1;
     }
     return Boolean.TRUE;
   }