Jackie-Jiang commented on code in PR #15261:
URL: https://github.com/apache/pinot/pull/15261#discussion_r2013034392
##########
pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/StreamIngestionConfig.java:
##########
@@ -43,6 +43,12 @@ public class StreamIngestionConfig extends BaseJsonConfig {
@JsonPropertyDescription("Whether pauseless consumption is enabled for the
table")
private boolean _pauselessConsumptionEnabled = false;
+ @JsonPropertyDescription("Enforce consumption of segments in order of
segment creation by the controller")
+ private boolean _enforceConsumptionInOrder = false;
+
+ @JsonPropertyDescription("If enabled, Server in-memory tracks the sequence
number of all loaded segments")
+ private boolean _trackSegmentSeqNumber = true;
Review Comment:
Since we are using this to determine whether to use IS to calculate the
previous segment, suggest making it more explicit (such as
_useIdealStateToCalculatePreviousSegment`). It is good practice to have boolean
config as `false` by default, which is less error-prone
##########
pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/StreamIngestionConfig.java:
##########
@@ -43,6 +43,12 @@ public class StreamIngestionConfig extends BaseJsonConfig {
@JsonPropertyDescription("Whether pauseless consumption is enabled for the
table")
private boolean _pauselessConsumptionEnabled = false;
+ @JsonPropertyDescription("Enforce consumption of segments in order of
segment creation by the controller")
+ private boolean _enforceConsumptionInOrder = false;
Review Comment:
This comment is not addressed
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinator.java:
##########
@@ -0,0 +1,234 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.common.utils.helix.HelixHelper;
+import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
+import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class ConsumerCoordinator {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ConsumerCoordinator.class);
+
+ private final Semaphore _semaphore;
+ private final boolean _enforceConsumptionInOrder;
+ private final Condition _condition;
+ private final Lock _lock;
+ private int _maxSegmentSeqNumLoaded = -1;
+ private boolean _relyOnIdealState = true;
Review Comment:
I see. I mis-read the logic and thought it is used only for the first
transition.
(minor) We can change it to `final`
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinator.java:
##########
@@ -0,0 +1,240 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import javax.annotation.Nullable;
+import org.apache.helix.model.IdealState;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.common.utils.helix.HelixHelper;
+import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
+import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The ConsumerCoordinator coordinates the offline->consuming helix
transitions.
+ */
+public class ConsumerCoordinator {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ConsumerCoordinator.class);
+
+ private final Semaphore _semaphore;
+ private final boolean _enforceConsumptionInOrder;
+ private final Condition _condition;
+ private final Lock _lock;
+ private int _maxSegmentSeqNumLoaded = -1;
Review Comment:
This needs to be `volatile`
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinator.java:
##########
@@ -0,0 +1,240 @@
+/**
+ * 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.core.data.manager.realtime;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import javax.annotation.Nullable;
+import org.apache.helix.model.IdealState;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.common.utils.helix.HelixHelper;
+import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
+import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The ConsumerCoordinator coordinates the offline->consuming helix
transitions.
+ */
+public class ConsumerCoordinator {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ConsumerCoordinator.class);
+
+ private final Semaphore _semaphore;
+ private final boolean _enforceConsumptionInOrder;
+ private final Condition _condition;
+ private final Lock _lock;
+ private int _maxSegmentSeqNumLoaded = -1;
+ private boolean _relyOnIdealState = true;
+ private final RealtimeTableDataManager _realtimeTableDataManager;
+ private final AtomicBoolean _isFirstTransitionProcessed;
+ private static final long WAIT_INTERVAL_MS = TimeUnit.MINUTES.toMillis(5);
+
+ public ConsumerCoordinator(boolean enforceConsumptionInOrder,
RealtimeTableDataManager realtimeTableDataManager) {
+ _semaphore = new Semaphore(1);
+ _lock = new ReentrantLock();
+ _condition = _lock.newCondition();
+ _enforceConsumptionInOrder = enforceConsumptionInOrder;
+ _realtimeTableDataManager = realtimeTableDataManager;
+ StreamIngestionConfig streamIngestionConfig =
realtimeTableDataManager.getStreamIngestionConfig();
+ boolean trackSegmentSeqNumber = false;
+ if (streamIngestionConfig != null) {
+ trackSegmentSeqNumber = streamIngestionConfig.isTrackSegmentSeqNumber();
+ }
+ // if trackSegmentSeqNumber is false, server relies on ideal state to
fetch previous segment to a segment.
+ if (trackSegmentSeqNumber) {
+ _relyOnIdealState = false;
+ }
+ _isFirstTransitionProcessed = new AtomicBoolean(false);
+ }
+
+ void acquire(LLCSegmentName llcSegmentName)
+ throws InterruptedException {
+
+ if (_enforceConsumptionInOrder) {
+ waitForPrevSegment(llcSegmentName);
+ }
+
+ long startTimeMs = System.currentTimeMillis();
+ while (!_semaphore.tryAcquire(WAIT_INTERVAL_MS, TimeUnit.MILLISECONDS)) {
+ LOGGER.warn("Failed to acquire partitionGroup consumer semaphore in: {}
ms. Retrying.",
+ System.currentTimeMillis() - startTimeMs);
+ }
+ }
+
+ void release() {
+ _semaphore.release();
+ }
+
+ @VisibleForTesting
+ Semaphore getSemaphore() {
+ return _semaphore;
+ }
+
+ void trackSegment(LLCSegmentName llcSegmentName) {
+ _lock.lock();
+ try {
+ if (!_relyOnIdealState) {
+ _maxSegmentSeqNumLoaded = Math.max(_maxSegmentSeqNumLoaded,
llcSegmentName.getSequenceNumber());
+ }
+ // notify all helix threads waiting for their offline -> consuming
segment's prev segment to be loaded
+ _condition.signalAll();
+ } finally {
+ _lock.unlock();
+ }
+ }
+
+ private void waitForPrevSegment(LLCSegmentName currSegment)
+ throws InterruptedException {
+
+ long startTimeMs = System.currentTimeMillis();
+
+ if ((_relyOnIdealState) || (!_isFirstTransitionProcessed.get())) {
Review Comment:
(nit) parenthesis seems redundant
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
