Offset auto reset

Author: gaoxin

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

@@ -29,7 +29,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();
 
@@ -38,25 +38,59 @@ 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);
+    Preconditions.checkArgument(topicName == null || !topicName.contains(SEPARATOR),
+        "Illegal topic 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;
+    _segmentName = segmentName;
   }
 
   /**
@@ -65,6 +99,10 @@ public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber
    */
   @Nullable
   public static LLCSegmentName of(String segmentName) {
+    String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
+    if (parts.length < 4 || parts.length > 5) {
+      return null;
+    }
     try {
       return new LLCSegmentName(segmentName);
     } catch (Exception e) {
@@ -76,13 +114,7 @@ public static LLCSegmentName of(String 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
@@ -94,17 +126,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;
   }
@@ -127,6 +176,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-common/src/test/java/org/apache/pinot/common/utils/LLCSegmentNameTest.java

@@ -42,6 +42,15 @@ public void testSegmentNameBuilder() {
     assertEquals(llcSegmentName.getPartitionGroupId(), 0);
     assertEquals(llcSegmentName.getSequenceNumber(), 1);
 
+    LLCSegmentName llcSegmentNameWithTopicName = new LLCSegmentName("myTable", "myTopic", 0, 1, 1465508537069L);
+    String segmentNameWithTopicName = llcSegmentNameWithTopicName.getSegmentName();
+    assertEquals(segmentNameWithTopicName, "myTable__myTopic__0__1__20160609T2142Z");
+    assertTrue(LLCSegmentName.isLLCSegment(segmentName));
+    assertEquals(llcSegmentNameWithTopicName.getTableName(), "myTable");
+    assertEquals(llcSegmentNameWithTopicName.getTopicName(), "myTopic");
+    assertEquals(llcSegmentNameWithTopicName.getPartitionGroupId(), 0);
+    assertEquals(llcSegmentNameWithTopicName.getSequenceNumber(), 1);
+
     // Invalid segment name
     assertFalse(LLCSegmentName.isLLCSegment("a__abc__1__3__4__54__g__gg___h"));
   }
@@ -95,4 +104,61 @@ public void testLLCSegmentName() {
     Arrays.sort(testSorted);
     Assert.assertEquals(testSorted, new LLCSegmentName[]{segName5, segName1, segName6, segName3, segName4});
   }
+
+  @Test
+  public void testLLCSegmentNameWithTopicName() {
+    String tableName = "myTable";
+    String topicName = "myTopic";
+    final int partitionGroupId = 4;
+    final int sequenceNumber = 27;
+    final long msSinceEpoch = 1466200248000L;
+    final String creationTime = "20160617T2150Z";
+    final long creationTimeInMs = 1466200200000L;
+    final String segmentName = "myTable__myTopic__4__27__" + creationTime;
+
+    LLCSegmentName segName1 = new LLCSegmentName(tableName, topicName, partitionGroupId, sequenceNumber, msSinceEpoch);
+    Assert.assertEquals(segName1.getSegmentName(), segmentName);
+    Assert.assertEquals(segName1.getPartitionGroupId(), partitionGroupId);
+    Assert.assertEquals(segName1.getCreationTime(), creationTime);
+    Assert.assertEquals(segName1.getCreationTimeMs(), creationTimeInMs);
+    Assert.assertEquals(segName1.getSequenceNumber(), sequenceNumber);
+    Assert.assertEquals(segName1.getTableName(), tableName);
+    Assert.assertEquals(segName1.getTopicName(), topicName);
+
+    LLCSegmentName segName2 = new LLCSegmentName(segmentName);
+    Assert.assertEquals(segName2.getSegmentName(), segmentName);
+    Assert.assertEquals(segName2.getPartitionGroupId(), partitionGroupId);
+    Assert.assertEquals(segName2.getCreationTime(), creationTime);
+    Assert.assertEquals(segName2.getCreationTimeMs(), creationTimeInMs);
+    Assert.assertEquals(segName2.getSequenceNumber(), sequenceNumber);
+    Assert.assertEquals(segName2.getTableName(), tableName);
+    Assert.assertEquals(segName2.getTopicName(), topicName);
+
+    Assert.assertEquals(segName1, segName2);
+
+    LLCSegmentName segName3 =
+        new LLCSegmentName(tableName, topicName, partitionGroupId + 1, sequenceNumber - 1, msSinceEpoch);
+    Assert.assertTrue(segName1.compareTo(segName3) < 0);
+    LLCSegmentName segName4 =
+        new LLCSegmentName(tableName, topicName, partitionGroupId + 1, sequenceNumber + 1, msSinceEpoch);
+    Assert.assertTrue(segName1.compareTo(segName4) < 0);
+    LLCSegmentName segName5 =
+        new LLCSegmentName(tableName, topicName, partitionGroupId - 1, sequenceNumber + 1, msSinceEpoch);
+    Assert.assertTrue(segName1.compareTo(segName5) > 0);
+    LLCSegmentName segName6 =
+        new LLCSegmentName(tableName, topicName, partitionGroupId, sequenceNumber + 1, msSinceEpoch);
+    Assert.assertTrue(segName1.compareTo(segName6) < 0);
+
+    LLCSegmentName segName7 =
+        new LLCSegmentName(tableName + "NotGood", topicName, partitionGroupId, sequenceNumber + 1, msSinceEpoch);
+    try {
+      segName1.compareTo(segName7);
+      Assert.fail("Not failing when comparing " + segName1.getSegmentName() + " and " + segName7.getSegmentName());
+    } catch (Exception e) {
+      // expected
+    }
+    LLCSegmentName[] testSorted = new LLCSegmentName[]{segName3, segName1, segName4, segName5, segName6};
+    Arrays.sort(testSorted);
+    Assert.assertEquals(testSorted, new LLCSegmentName[]{segName5, segName1, segName6, segName3, segName4});
+  }
 }

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

@@ -118,6 +118,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;
@@ -190,6 +191,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;
@@ -385,6 +387,10 @@ public RealtimeSegmentValidationManager getRealtimeSegmentValidationManager() {
     return _realtimeSegmentValidationManager;
   }
 
+  public RealtimeOffsetAutoResetManager getRealtimeOffsetAutoResetManager() {
+    return _realtimeOffsetAutoResetManager;
+  }
+
   public BrokerResourceValidationManager getBrokerResourceValidationManager() {
     return _brokerResourceValidationManager;
   }
@@ -880,6 +886,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

@@ -114,6 +114,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 =
@@ -272,6 +276,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
@@ -674,6 +679,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
@@ -1153,6 +1168,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

@@ -54,8 +54,13 @@ private Constants() {
   public static final String PERIODIC_TASK_TAG = "PeriodicTask";
   public static final String UPSERT_RESOURCE_TAG = "Upsert";
   public static final String QUERY_WORKLOAD_TAG = "QueryWorkload";
+  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

@@ -2106,6 +2106,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,59 @@
+/**
+ * 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);
+
+  public abstract void close();
+}