Offset auto reset

Author: gaoxin

pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java

@@ -28,7 +28,7 @@
 
 
 public class LLCSegmentName implements Comparable<LLCSegmentName> {
-  private static final String SEPARATOR = "__";
+  public static final String SEPARATOR = "__";
   private static final String DATE_FORMAT = "yyyyMMdd'T'HHmm'Z'";
   private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormat.forPattern(DATE_FORMAT).withZoneUTC();
 
@@ -37,30 +37,53 @@ public class LLCSegmentName implements Comparable<LLCSegmentName> {
   private final int _sequenceNumber;
   private final String _creationTime;
   private final String _segmentName;
+  private final String _topicName;
 
   public LLCSegmentName(String segmentName) {
     String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
-    Preconditions.checkArgument(parts.length == 4, "Invalid LLC segment name: %s", segmentName);
+    Preconditions.checkArgument(
+        parts.length >= 4 && parts.length <= 5, "Invalid LLC segment name: %s", segmentName);
     _tableName = parts[0];
-    _partitionGroupId = Integer.parseInt(parts[1]);
-    _sequenceNumber = Integer.parseInt(parts[2]);
-    _creationTime = parts[3];
+    if (parts.length == 4) {
+      _topicName = "";
+      _partitionGroupId = Integer.parseInt(parts[1]);
+      _sequenceNumber = Integer.parseInt(parts[2]);
+      _creationTime = parts[3];
+    } else {
+      _topicName = parts[1];
+      _partitionGroupId = Integer.parseInt(parts[2]);
+      _sequenceNumber = Integer.parseInt(parts[3]);
+      _creationTime = parts[4];
+    }
     _segmentName = segmentName;
   }
 
   public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
+    this(tableName, "", partitionGroupId, sequenceNumber, msSinceEpoch);
+  }
+
+  public LLCSegmentName(
+      String tableName, String topicName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
     Preconditions.checkArgument(!tableName.contains(SEPARATOR), "Illegal table name: %s", tableName);
     _tableName = tableName;
+    _topicName = topicName;
     _partitionGroupId = partitionGroupId;
     _sequenceNumber = sequenceNumber;
     // ISO8601 date: 20160120T1234Z
     _creationTime = DATE_FORMATTER.print(msSinceEpoch);
-    _segmentName = tableName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR + _creationTime;
+    if ("".equals(topicName)) {
+      _segmentName = tableName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR + _creationTime;
+    } else {
+      _segmentName =
+          tableName + SEPARATOR + topicName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR
+              + _creationTime;
+    }
   }
 
   private LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, String creationTime,
       String segmentName) {
     _tableName = tableName;
+    _topicName = "";
     _partitionGroupId = partitionGroupId;
     _sequenceNumber = sequenceNumber;
     _creationTime = creationTime;
@@ -74,23 +97,17 @@ private LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumbe
   @Nullable
   public static LLCSegmentName of(String segmentName) {
     String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
-    if (parts.length != 4) {
+    if (parts.length < 4 || parts.length > 5) {
       return null;
     }
-    return new LLCSegmentName(parts[0], Integer.parseInt(parts[1]), Integer.parseInt(parts[2]), parts[3], segmentName);
+    return new LLCSegmentName(segmentName);
   }
 
   /**
    * Returns whether the given segment name represents an LLC segment.
    */
   public static boolean isLLCSegment(String segmentName) {
-    int numSeparators = 0;
-    int index = 0;
-    while ((index = segmentName.indexOf(SEPARATOR, index)) != -1) {
-      numSeparators++;
-      index += 2; // SEPARATOR.length()
-    }
-    return numSeparators == 3;
+    return of(segmentName) != null;
   }
 
   @Deprecated
@@ -102,17 +119,34 @@ public static boolean isLowLevelConsumerSegmentName(String segmentName) {
    * Returns the sequence number of the given segment name.
    */
   public static int getSequenceNumber(String segmentName) {
-    return Integer.parseInt(StringUtils.splitByWholeSeparator(segmentName, SEPARATOR)[2]);
+    String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
+    if (parts.length == 4) {
+      return Integer.parseInt(parts[2]);
+    } else {
+      return Integer.parseInt(parts[3]);
+    }
   }
 
   public String getTableName() {
     return _tableName;
   }
 
+  public String getTopicName() {
+    return _topicName;
+  }
+
   public int getPartitionGroupId() {
     return _partitionGroupId;
   }
 
+  public String getPartitionGroupInfo() {
+    if (_topicName.isEmpty()) {
+      return String.valueOf(_partitionGroupId);
+    } else {
+      return _topicName + SEPARATOR + _partitionGroupId;
+    }
+  }
+
   public int getSequenceNumber() {
     return _sequenceNumber;
   }
@@ -134,6 +168,9 @@ public String getSegmentName() {
   public int compareTo(LLCSegmentName other) {
     Preconditions.checkArgument(_tableName.equals(other._tableName),
         "Cannot compare segment names from different table: %s, %s", _segmentName, other.getSegmentName());
+    if (!_topicName.equals(other._topicName)) {
+      return StringUtils.compare(_topicName, other._topicName);
+    }
     if (_partitionGroupId != other._partitionGroupId) {
       return Integer.compare(_partitionGroupId, other._partitionGroupId);
     }

pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java

@@ -114,6 +114,7 @@
 import org.apache.pinot.controller.validation.BrokerResourceValidationManager;
 import org.apache.pinot.controller.validation.DiskUtilizationChecker;
 import org.apache.pinot.controller.validation.OfflineSegmentIntervalChecker;
+import org.apache.pinot.controller.validation.RealtimeOffsetAutoResetManager;
 import org.apache.pinot.controller.validation.RealtimeSegmentValidationManager;
 import org.apache.pinot.controller.validation.ResourceUtilizationChecker;
 import org.apache.pinot.controller.validation.ResourceUtilizationManager;
@@ -185,6 +186,7 @@ public abstract class BaseControllerStarter implements ServiceStartable {
   // Can only be constructed after resource manager getting started
   protected OfflineSegmentIntervalChecker _offlineSegmentIntervalChecker;
   protected RealtimeSegmentValidationManager _realtimeSegmentValidationManager;
+  protected RealtimeOffsetAutoResetManager _realtimeOffsetAutoResetManager;
   protected BrokerResourceValidationManager _brokerResourceValidationManager;
   protected SegmentRelocator _segmentRelocator;
   protected RetentionManager _retentionManager;
@@ -373,6 +375,10 @@ public RealtimeSegmentValidationManager getRealtimeSegmentValidationManager() {
     return _realtimeSegmentValidationManager;
   }
 
+  public RealtimeOffsetAutoResetManager getRealtimeOffsetAutoResetManager() {
+    return _realtimeOffsetAutoResetManager;
+  }
+
   public BrokerResourceValidationManager getBrokerResourceValidationManager() {
     return _brokerResourceValidationManager;
   }
@@ -850,6 +856,9 @@ protected List<PeriodicTask> setupControllerPeriodicTasks() {
             _pinotLLCRealtimeSegmentManager, _validationMetrics, _controllerMetrics, _storageQuotaChecker,
             _resourceUtilizationManager);
     periodicTasks.add(_realtimeSegmentValidationManager);
+    _realtimeOffsetAutoResetManager =
+        new RealtimeOffsetAutoResetManager(_config, _helixResourceManager, _leadControllerManager,
+            _pinotLLCRealtimeSegmentManager, _controllerMetrics);
     _brokerResourceValidationManager =
         new BrokerResourceValidationManager(_config, _helixResourceManager, _leadControllerManager, _controllerMetrics);
     periodicTasks.add(_brokerResourceValidationManager);

pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java

@@ -112,6 +112,10 @@ public static class ControllerPeriodicTasksConf {
         "controller.realtime.segment.validation.frequencyPeriod";
     public static final String REALTIME_SEGMENT_VALIDATION_INITIAL_DELAY_IN_SECONDS =
         "controller.realtime.segment.validation.initialDelayInSeconds";
+    public static final String REALTIME_OFFSET_AUTO_RESET_FREQUENCY_PERIOD =
+        "controller.realtime.offsetAutoReset.frequencyPeriod";
+    public static final String REALTIME_OFFSET_AUTO_RESET_INITIAL_DELAY_IN_SECONDS =
+        "controller.realtime.offsetAutoReset.initialDelayInSeconds";
     // Deprecated as of 0.8.0
     @Deprecated
     public static final String DEPRECATED_BROKER_RESOURCE_VALIDATION_FREQUENCY_IN_SECONDS =
@@ -268,6 +272,7 @@ public static long getRandomInitialDelayInSeconds() {
     public static final int DEFAULT_RETENTION_MANAGER_FREQUENCY_IN_SECONDS = 6 * 60 * 60; // 6 Hours.
     public static final int DEFAULT_OFFLINE_SEGMENT_INTERVAL_CHECKER_FREQUENCY_IN_SECONDS = 24 * 60 * 60; // 24 Hours.
     public static final int DEFAULT_REALTIME_SEGMENT_VALIDATION_FREQUENCY_IN_SECONDS = 60 * 60; // 1 Hour.
+    public static final int DEFAULT_OFFSET_AUTO_RESET_FREQUENCY_IN_SECONDS = 60 * 60; // 1 Hour.
     public static final int DEFAULT_BROKER_RESOURCE_VALIDATION_FREQUENCY_IN_SECONDS = 60 * 60; // 1 Hour.
     public static final int DEFAULT_STATUS_CHECKER_FREQUENCY_IN_SECONDS = 5 * 60; // 5 minutes
     public static final int DEFAULT_REBALANCE_CHECKER_FREQUENCY_IN_SECONDS = 5 * 60; // 5 minutes
@@ -657,6 +662,16 @@ public void setRealtimeSegmentValidationFrequencyInSeconds(int validationFrequen
         Integer.toString(validationFrequencyInSeconds));
   }
 
+  public int getRealtimeOffsetAutoResetFrequencyInSeconds() {
+    return getProperty(ControllerPeriodicTasksConf.REALTIME_OFFSET_AUTO_RESET_FREQUENCY_PERIOD,
+        ControllerPeriodicTasksConf.DEFAULT_OFFSET_AUTO_RESET_FREQUENCY_IN_SECONDS);
+  }
+
+  public void setRealtimeOffsetAutoResetFrequencyInSeconds(int offsetAutoResetFrequencyInSeconds) {
+    setProperty(ControllerPeriodicTasksConf.REALTIME_OFFSET_AUTO_RESET_FREQUENCY_PERIOD,
+        Integer.toString(offsetAutoResetFrequencyInSeconds));
+  }
+
   /**
    * Return <code>controller.broker.resource.validation.frequencyPeriod</code> or
    * <code>controller.broker.resource.validation.frequencyInSeconds</code> or the default broker resource validation
@@ -1107,6 +1122,11 @@ public long getRealtimeSegmentValidationManagerInitialDelaySeconds() {
         getPeriodicTaskInitialDelayInSeconds());
   }
 
+  public long getRealtimeOffsetAutoResetInitialDelaySeconds() {
+    return getProperty(ControllerPeriodicTasksConf.REALTIME_OFFSET_AUTO_RESET_INITIAL_DELAY_IN_SECONDS,
+        getPeriodicTaskInitialDelayInSeconds());
+  }
+
   public boolean isDeepStoreRetryUploadLLCSegmentEnabled() {
     return getProperty(ControllerPeriodicTasksConf.ENABLE_DEEP_STORE_RETRY_UPLOAD_LLC_SEGMENT, false);
   }

pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/Constants.java

@@ -53,8 +53,13 @@ private Constants() {
   public static final String APP_CONFIGS = "AppConfigs";
   public static final String PERIODIC_TASK_TAG = "PeriodicTask";
   public static final String UPSERT_RESOURCE_TAG = "Upsert";
+  public static final String RESET_OFFSET_FROM = "ResetOffsetFrom";
+  public static final String RESET_OFFSET_TO = "ResetOffsetTo";
+  public static final String RESET_OFFSET_TOPIC_NAME = "ResetOffsetTopicName";
+  public static final String RESET_OFFSET_TOPIC_PARTITION = "ResetOffsetTopicPartition";
 
   public static final String REALTIME_SEGMENT_VALIDATION_MANAGER = "RealtimeSegmentValidationManager";
+  public static final String REALTIME_OFFSET_AUTO_RESET_MANAGER = "RealtimeOffsetAutoResetManager";
 
   public static TableType validateTableType(String tableTypeStr) {
     if (StringUtils.isBlank(tableTypeStr)) {

pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java

@@ -2038,6 +2038,7 @@ public void updateTableConfig(TableConfig tableConfig)
       throws IOException {
     validateTableTenantConfig(tableConfig);
     validateTableTaskMinionInstanceTagConfig(tableConfig);
+    // TODO: make sure external table update would not remove the ephemeral backfill topics
     setExistingTableConfig(tableConfig);
   }
 

pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/RealtimeOffsetAutoResetHandler.java

@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.helix.core.periodictask;
+
+import java.util.Collection;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.realtime.PinotLLCRealtimeSegmentManager;
+import org.apache.pinot.spi.stream.StreamConfig;
+
+
+public abstract class RealtimeOffsetAutoResetHandler {
+  protected PinotLLCRealtimeSegmentManager _llcRealtimeSegmentManager;
+  protected PinotHelixResourceManager _pinotHelixResourceManager;
+
+  public RealtimeOffsetAutoResetHandler(PinotLLCRealtimeSegmentManager llcRealtimeSegmentManager,
+      PinotHelixResourceManager pinotHelixResourceManager) {
+    _llcRealtimeSegmentManager = llcRealtimeSegmentManager;
+    _pinotHelixResourceManager = pinotHelixResourceManager;
+  }
+
+  /**
+   * Trigger the job to backfill the skipped interval due to offset auto reset.
+   * It is expected to backfill the [fromOffset, toOffset) interval.
+   * @return if successfully started the backfill job
+   */
+  public abstract boolean triggerBackfillJob(
+      String tableNameWithType, StreamConfig streamConfig, String topicName, int partitionId, long fromOffset,
+      long toOffset);
+
+  /**
+   * Ensure all topics under the table are being backfilled. It is the caller's responsibility to figure out what
+   * topic set it should check.
+   */
+  public abstract void ensureBackfillJobsRunning(String tableNameWithType, Collection<String> topicNames);
+
+  /**
+   * Return the Collection of completed and cleaned up topicNames from the input.
+   */
+  public abstract Collection<String> cleanupCompletedBackfillJobs(
+      String tableNameWithType, Collection<String> topicNames);
+}