Fix issues, rebase and resolve comments

apache · Oct 17, 2024 · 1af31b2 · 1af31b2
1 parent 48dc8dc
commit 1af31b2
Show file tree

Hide file tree

Showing 5 changed files with 16 additions and 5 deletions.
diff --git a/...rc/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java b/...rc/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
@@ -195,7 +195,7 @@ static long computeTotalDocumentCount(List<SegmentZKMetadata> segmentsZKMetadata
   @Override
   public void cleanUpTask() {
     LOGGER.info("Unregister all the validation metrics.");
-    _validationMetrics.unregisterAllMetrics();/PinotLLCRealtimeSegmentManagerTest
+    _validationMetrics.unregisterAllMetrics();
   }
 
   public static final class Context {

diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
@@ -174,6 +174,10 @@ public static void validate(TableConfig tableConfig, @Nullable Schema schema, @N
       // Only allow realtime tables with non-null stream.type and LLC consumer.type
       if (tableConfig.getTableType() == TableType.REALTIME) {
         List<Map<String, String>> streamConfigMaps = IngestionConfigUtils.getStreamConfigMaps(tableConfig);
+        if (streamConfigMaps.size() > 1) {
+          Preconditions.checkArgument(!tableConfig.isUpsertEnabled(),
+              "Multiple stream configs are not supported for upsert tables");
+        }
         // TODO: validate stream configs in the map are identical in most fields
         StreamConfig streamConfig;
         for (Map<String, String> streamConfigMap : streamConfigMaps) {

diff --git a/...egment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java b/...egment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
@@ -685,7 +685,7 @@ public void ingestionStreamConfigsTest() {
         new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName("timeColumn")
             .setIngestionConfig(ingestionConfig).build();
 
-    // only 1 stream config allowed
+    // Multiple stream configs is allowed
     try {
       TableConfigUtils.validateIngestionConfig(tableConfig, null);
     } catch (IllegalStateException e) {

diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java
@@ -74,17 +74,24 @@ public Exception getException() {
   @Override
   public Boolean call()
       throws Exception {
+    _newPartitionGroupMetadataList.clear();
     for (int i = 0; i < _streamConfigs.size(); i++) {
       String clientId = PartitionGroupMetadataFetcher.class.getSimpleName() + "-"
           + _streamConfigs.get(i).getTableNameWithType() + "-" + _topicNames.get(i);
       StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(_streamConfigs.get(i));
+      final int index = i;
+      List<PartitionGroupConsumptionStatus> topicPartitionGroupConsumptionStatusList =
+          _partitionGroupConsumptionStatusList.stream()
+              .filter(partitionGroupConsumptionStatus ->
+                  IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
+                      partitionGroupConsumptionStatus.getPartitionGroupId()) == index)
+              .collect(Collectors.toList());
       try (
           StreamMetadataProvider streamMetadataProvider =
               streamConsumerFactory.createStreamMetadataProvider(clientId)) {
-        final int index = i;
         _newPartitionGroupMetadataList.addAll(streamMetadataProvider.computePartitionGroupMetadata(clientId,
             _streamConfigs.get(i),
-            _partitionGroupConsumptionStatusList, /*maxWaitTimeMs=*/5000).stream().map(
+            topicPartitionGroupConsumptionStatusList, /*maxWaitTimeMs=*/5000).stream().map(
                 metadata -> new PartitionGroupMetadata(
                     IngestionConfigUtils.getPinotPartitionIdFromStreamPartitionId(
                         metadata.getPartitionGroupId(), index),

diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java
@@ -81,7 +81,7 @@ default List<PartitionGroupMetadata> computePartitionGroupMetadata(String client
     // If partition group is still in progress, this value will be null
     for (PartitionGroupConsumptionStatus currentPartitionGroupConsumptionStatus : partitionGroupConsumptionStatuses) {
       newPartitionGroupMetadataList.add(
-          new PartitionGroupMetadata(currentPartitionGroupConsumptionStatus.getPartitionGroupId(),
+          new PartitionGroupMetadata(currentPartitionGroupConsumptionStatus.getStreamPartitionGroupId(),
               currentPartitionGroupConsumptionStatus.getEndOffset()));
     }
     // Add PartitionGroupMetadata for new partitions