1. Removing pauseless from indexingConfig

2. Fixing few formatting issues. 3. Fixing few comments for better readability. 4. Removing unused code: contructors, constants etc
apache · Jan 7, 2025 · 4f92b6d · 4f92b6d
1 parent a97847f
commit 4f92b6d
Show file tree

Hide file tree

Showing 5 changed files with 3 additions and 29 deletions.
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/PauselessConsumptionUtils.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/PauselessConsumptionUtils.java
@@ -20,14 +20,12 @@
 
 import java.util.Optional;
 import javax.validation.constraints.NotNull;
-import org.apache.pinot.spi.config.table.IndexingConfig;
 import org.apache.pinot.spi.config.table.TableConfig;
 import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
 import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
 
 
 public class PauselessConsumptionUtils {
-  public static final String PAUSELESS_CONSUMPTION_ENABLED = "pauselessConsumptionEnabled";
 
   private PauselessConsumptionUtils() {
     // Private constructor to prevent instantiation of utility class
@@ -42,15 +40,6 @@ private PauselessConsumptionUtils() {
    * @throws NullPointerException if tableConfig is null
    */
   public static boolean isPauselessEnabled(@NotNull TableConfig tableConfig) {
-    return checkIngestionConfig(tableConfig) || checkIndexingConfig(tableConfig);
-  }
-
-  private static boolean checkIndexingConfig(@NotNull TableConfig tableConfig) {
-    return Optional.ofNullable(tableConfig.getIndexingConfig()).map(IndexingConfig::isPauselessConsumptionEnabled)
-        .orElse(false);
-  }
-
-  private static boolean checkIngestionConfig(@NotNull TableConfig tableConfig) {
     return Optional.ofNullable(tableConfig.getIngestionConfig()).map(IngestionConfig::getStreamIngestionConfig)
         .map(StreamIngestionConfig::isPauselessConsumptionEnabled).orElse(false);
   }

diff --git a/.../src/main/java/org/apache/pinot/core/data/manager/realtime/PauselessSegmentCommitter.java b/.../src/main/java/org/apache/pinot/core/data/manager/realtime/PauselessSegmentCommitter.java
@@ -32,11 +32,6 @@ public PauselessSegmentCommitter(Logger segmentLogger, ServerSegmentCompletionPr
     super(segmentLogger, protocolHandler, params, segmentUploader, peerDownloadScheme);
   }
 
-  public PauselessSegmentCommitter(Logger segmentLogger, ServerSegmentCompletionProtocolHandler protocolHandler,
-      SegmentCompletionProtocol.Request.Params params, SegmentUploader segmentUploader) {
-    super(segmentLogger, protocolHandler, params, segmentUploader);
-  }
-
   /**
    * Commits a built segment without executing the segmentCommitStart step. This method assumes that
    * segmentCommitStart has already been executed prior to building the segment.

diff --git a/...src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/...src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
@@ -851,8 +851,7 @@ public void run() {
                 // 2. Creates ZK metadata for the new consuming segment
                 // 3. Updates the IdealState for committing and new consuming segment to ONLINE and CONSUMING
                 // respectively.
-                // See design doc for the new commit protocol:
-                // https://docs.google.com/document/d/1d-xttk7sXFIOqfyZvYw5W_KeGS6Ztmi8eYevBcCrT_c
+                // Refer to the PR for the new commit protocol: https://github.com/apache/pinot/pull/14741
                 if (PauselessConsumptionUtils.isPauselessEnabled(_tableConfig)) {
                   if (!startSegmentCommit()) {
                     // If for any reason commit failed, we don't want to be in COMMITTING state when we hold.
@@ -925,7 +924,7 @@ public void run() {
     }
   }
 
-  boolean startSegmentCommit() {
+  private boolean startSegmentCommit() {
     SegmentCompletionProtocol.Request.Params params = new SegmentCompletionProtocol.Request.Params();
     params.withSegmentName(_segmentNameStr).withStreamPartitionMsgOffset(_currentOffset.toString())
         .withNumRows(_numRowsConsumed).withInstanceId(_instanceId).withReason(_stopReason);

diff --git a/...e/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java b/...e/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
@@ -601,7 +601,7 @@ public File downloadSegment(SegmentZKMetadata zkMetadata)
       Thread.sleep(Math.min(SLEEP_INTERVAL_MS, timeRemaining));
     }
 
-// If we exit the loop without returning, throw an exception
+    // If we exit the loop without returning, throw an exception
     throw new TimeoutException("Failed to download segment after " + TIMEOUT_MINUTES + " minutes of retrying. Segment: "
         + zkMetadata.getSegmentName());
   }

diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java
@@ -66,7 +66,6 @@ public class IndexingConfig extends BaseJsonConfig {
   private boolean _aggregateMetrics;
   private boolean _nullHandlingEnabled;
   private boolean _columnMajorSegmentBuilderEnabled = true;
-  private boolean _pauselessConsumptionEnabled = false;
 
   /**
    * If `optimizeDictionary` enabled, dictionary is not created for the high-cardinality
@@ -460,12 +459,4 @@ public Set<String> getAllReferencedColumns() {
     }
     return allColumns;
   }
-
-  public boolean isPauselessConsumptionEnabled() {
-    return _pauselessConsumptionEnabled;
-  }
-
-  public void setPauselessConsumptionEnabled(boolean pauselessConsumptionEnabled) {
-    _pauselessConsumptionEnabled = pauselessConsumptionEnabled;
-  }
 }