You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "KKcorps (via GitHub)" <gi...@apache.org> on 2024/09/26 09:32:53 UTC
[PR] [WIP] Pluggable fsm [pinot]
KKcorps opened a new pull request, #14088:
URL: https://github.com/apache/pinot/pull/14088
DO NOT MERGE THIS
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [WIP] Pluggable fsm [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #14088:
URL: https://github.com/apache/pinot/pull/14088#issuecomment-2376539009
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/14088?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 55.39%. Comparing base [(`59551e4`)](https://app.codecov.io/gh/apache/pinot/commit/59551e45224f1535c4863fd577622b37366ccc97?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`3e68850`)](https://app.codecov.io/gh/apache/pinot/commit/3e68850a3dbccf8c3d1a015ebf16b85f7709c189?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1091 commits behind head on master.
> :exclamation: There is a different number of reports uploaded between BASE (59551e4) and HEAD (3e68850). Click for more details.
>
> <details><summary>HEAD has 53 uploads less than BASE</summary>
>
>| Flag | BASE (59551e4) | HEAD (3e68850) |
>|------|------|------|
>|integration|7|0|
>|integration2|3|0|
>|temurin|12|1|
>|java-21|7|1|
>|skip-bytebuffers-true|3|0|
>|skip-bytebuffers-false|7|1|
>|unittests|5|1|
>|unittests1|2|1|
>|java-11|5|0|
>|unittests2|3|0|
>|integration1|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #14088 +/- ##
============================================
- Coverage 61.75% 55.39% -6.36%
- Complexity 207 705 +498
============================================
Files 2436 2041 -395
Lines 133233 108217 -25016
Branches 20636 17168 -3468
============================================
- Hits 82274 59947 -22327
+ Misses 44911 43467 -1444
+ Partials 6048 4803 -1245
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.39% <ø> (-6.24%)` | :arrow_down: |
| [skip-bytebuffers-false](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.39% <ø> (-6.36%)` | :arrow_down: |
| [skip-bytebuffers-true](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.39% <ø> (-6.36%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.39% <ø> (-6.36%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.39% <ø> (+8.50%)` | :arrow_up: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/14088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/14088?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Make segment completion FSM pluggable [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #14088:
URL: https://github.com/apache/pinot/pull/14088#discussion_r1880235825
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSM.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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.realtime;
+
+import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
+import org.apache.pinot.controller.helix.core.realtime.segment.CommittingSegmentDescriptor;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+
+
+/**
+ * Interface for managing the state machine transitions related to segment completion
+ * in a real-time table within Apache Pinot.
+ *
+ * The segment completion process is crucial for handling real-time ingestion, ensuring
+ * that segments are correctly built, committed, or discarded based on the server's state
+ * and the protocol defined for low-level consumer (LLC) tables. This interface abstracts
+ * the controller-side logic for responding to server events during the lifecycle of a
+ * segment's completion.
+ */
+public interface SegmentCompletionFSM {
+
+ /**
+ * Initializes the FSM to its initial state based on the message type.
+ * @param msgType
+ */
+ void transitionToInitialState(String msgType);
Review Comment:
@Jackie-Jiang added this new method to remove the msgType from constructor
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Make segment completion FSM pluggable [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #14088:
URL: https://github.com/apache/pinot/pull/14088#discussion_r1868218501
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionConfig.java:
##########
@@ -0,0 +1,44 @@
+/**
+ * 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.realtime;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.spi.env.PinotConfiguration;
+
+
+public class SegmentCompletionConfig {
+ private final Map<String, String> _fsmSchemes = new HashMap<>();
+
+ public SegmentCompletionConfig(PinotConfiguration configuration) {
Review Comment:
Seems not addressed, or you haven't pushed the local change?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSMState.java:
##########
@@ -0,0 +1,30 @@
+/**
+ * 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.realtime;
+
+public enum SegmentCompletionFSMState {
Review Comment:
In that case, should we make this an inner class if this is implementation specific?
Also the indentation is wrong
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSMFactory.java:
##########
@@ -0,0 +1,144 @@
+/**
+ * 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.realtime;
+
+import com.google.common.base.Preconditions;
+import java.lang.reflect.Constructor;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SegmentCompletionFSMFactory {
+ private SegmentCompletionFSMFactory() {
+ // Private constructor to prevent instantiation
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SegmentCompletionFSMFactory.class);
+ private static final Map<String, Constructor<? extends SegmentCompletionFSM>> FSM_CONSTRUCTOR_MAP = new HashMap<>();
+
+ // Static block to register the default FSM
+ static {
+ try {
+ Class<?> clazz = Class.forName("org.apache.pinot.controller.helix.core.realtime.BlockingSegmentCompletionFSM");
+ Constructor<?> constructor = clazz.getConstructor(
+ PinotLLCRealtimeSegmentManager.class,
+ SegmentCompletionManager.class,
+ LLCSegmentName.class,
+ SegmentZKMetadata.class,
+ String.class
Review Comment:
In that case, let's not put it in the constructor. We should have a method to forward the FSM to a certain state
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Make segment completion FSM pluggable [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #14088:
URL: https://github.com/apache/pinot/pull/14088#discussion_r1880187070
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionConfig.java:
##########
@@ -0,0 +1,44 @@
+/**
+ * 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.realtime;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.spi.env.PinotConfiguration;
+
+
+public class SegmentCompletionConfig {
+ private final Map<String, String> _fsmSchemes = new HashMap<>();
+
+ public SegmentCompletionConfig(PinotConfiguration configuration) {
Review Comment:
It is addressed right, the comment is on wrong line I belive
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Make segment completion FSM pluggable [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #14088:
URL: https://github.com/apache/pinot/pull/14088#discussion_r1882808910
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSMFactory.java:
##########
@@ -0,0 +1,132 @@
+/**
+ * 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.realtime;
+
+import com.google.common.base.Preconditions;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SegmentCompletionFSMFactory {
+ private SegmentCompletionFSMFactory() {
+ // Private constructor to prevent instantiation
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SegmentCompletionFSMFactory.class);
+ private static final Map<String, Class<? extends SegmentCompletionFSM>> FSM_CLASS_MAP = new HashMap<>();
+
+ // Static block to register the default FSM
+ static {
+ try {
+ Class<?> clazz = Class.forName(BlockingSegmentCompletionFSM.class.getCanonicalName());
+ register(SegmentCompletionConfig.DEFAULT_FSM_SCHEME, (Class<? extends SegmentCompletionFSM>) clazz);
+ LOGGER.info("Registered default BlockingSegmentCompletionFSM");
+ } catch (Exception e) {
+ LOGGER.error("Failed to register default BlockingSegmentCompletionFSM", e);
+ throw new RuntimeException("Failed to register default BlockingSegmentCompletionFSM", e);
+ }
+ }
+
+ /**
+ * Registers an FSM class with a specific scheme/type.
+ *
+ * @param scheme The scheme or type key.
+ * @param fsmClass The class for the FSM.
+ */
+ public static void register(String scheme, Class<? extends SegmentCompletionFSM> fsmClass) {
+ Preconditions.checkNotNull(scheme, "Scheme cannot be null");
+ Preconditions.checkNotNull(fsmClass, "FSM Class cannot be null");
+ if (FSM_CLASS_MAP.containsKey(scheme)) {
+ LOGGER.warn("Overwriting existing FSM class for scheme {}", scheme);
Review Comment:
Let's throw exception here to avoid unexpected registry:
`Preconditions.checkState(FSM_CLASS_MAP.put(scheme, fsmClass) == null, ...)`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSM.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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.realtime;
+
+import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
+import org.apache.pinot.controller.helix.core.realtime.segment.CommittingSegmentDescriptor;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+
+
+/**
+ * Interface for managing the state machine transitions related to segment completion
+ * in a real-time table within Apache Pinot.
+ *
+ * The segment completion process is crucial for handling real-time ingestion, ensuring
+ * that segments are correctly built, committed, or discarded based on the server's state
+ * and the protocol defined for low-level consumer (LLC) tables. This interface abstracts
+ * the controller-side logic for responding to server events during the lifecycle of a
+ * segment's completion.
+ */
+public interface SegmentCompletionFSM {
+
+ /**
+ * Initializes the FSM to its initial state based on the message type.
+ * @param msgType
+ */
+ void transitionToInitialState(String msgType);
+
+ /**
Review Comment:
(format) Reformat this
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/BlockingSegmentCompletionFSM.java:
##########
@@ -0,0 +1,905 @@
+/**
+ * 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.realtime;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
+import org.apache.pinot.common.utils.HashUtil;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.common.utils.URIUtils;
+import org.apache.pinot.controller.helix.core.realtime.segment.CommittingSegmentDescriptor;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffsetFactory;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class implements the FSM on the controller side for each completing segment.
+ *
+ * An FSM is is created when we first hear about a segment (typically through the segmentConsumed message).
+ * When an FSM is created, it may have one of two start states (HOLDING, or COMMITTED), depending on the
+ * constructor used.
+ *
+ * We kick off an FSM in the COMMITTED state (rare) when we find that PROPERTYSTORE already has the segment
+ * with the Status set to DONE.
+ *
+ * We kick off an FSM in the HOLDING state (typical) when a sementConsumed() message arrives from the
+ * first server we hear from.
+ *
+ * The FSM does not have a timer. It is clocked by the servers, which, typically, are retransmitting their
+ * segmentConsumed() message every so often (SegmentCompletionProtocol.MAX_HOLD_TIME_MS).
+ *
+ * See https://github.com/linkedin/pinot/wiki/Low-level-kafka-consumers
+ */
+public class BlockingSegmentCompletionFSM implements SegmentCompletionFSM {
+ public static final Logger LOGGER = LoggerFactory.getLogger(BlockingSegmentCompletionFSM.class);
+
+ public enum BlockingSegmentCompletionFSMState {
+ PARTIAL_CONSUMING, // Indicates that at least one replica has reported that it has stopped consuming.
+ HOLDING, // the segment has started finalizing.
+ COMMITTER_DECIDED, // We know who the committer will be, we will let them know next time they call segmentConsumed()
+ COMMITTER_NOTIFIED, // we notified the committer to commit.
+ COMMITTER_UPLOADING, // committer is uploading.
+ COMMITTING, // we are in the process of committing to zk
+ COMMITTED, // We already committed a segment.
+ ABORTED, // state machine is aborted. we will start a fresh one when the next segmentConsumed comes in.
+ }
+
+ // We will have some variation between hosts, so we add 10% to the max hold time to pick a winner.
+ // If there is more than 10% variation, then it is handled as an error case (i.e. the first few to
+ // come in will have a winner, and the later ones will just download the segment)
+ private static final long MAX_TIME_TO_PICK_WINNER_MS =
+ SegmentCompletionProtocol.MAX_HOLD_TIME_MS + (SegmentCompletionProtocol.MAX_HOLD_TIME_MS / 10);
+
+ // Once we pick a winner, the winner may get notified in the next call, so add one hold time plus some.
+ // It may be that the winner is not the server that we are currently processing a segmentConsumed()
+ // message from. In that case, we will wait for the next segmetnConsumed() message from the picked winner.
+ // If the winner does not come back to us within that time, we abort the state machine and start over.
+ private static final long MAX_TIME_TO_NOTIFY_WINNER_MS =
+ MAX_TIME_TO_PICK_WINNER_MS + SegmentCompletionProtocol.MAX_HOLD_TIME_MS + (
+ SegmentCompletionProtocol.MAX_HOLD_TIME_MS / 10);
+
+ public final Logger _logger;
+
+ BlockingSegmentCompletionFSMState _state = BlockingSegmentCompletionFSMState.HOLDING;
+ // Typically start off in HOLDING state.
+ final long _startTimeMs;
+ private final LLCSegmentName _segmentName;
+ private final String _rawTableName;
+ private final String _realtimeTableName;
+ private final int _numReplicas;
+ private final Set<String> _excludedServerStateMap;
+ private final Map<String, StreamPartitionMsgOffset> _commitStateMap;
+ private final StreamPartitionMsgOffsetFactory _streamPartitionMsgOffsetFactory;
+ private StreamPartitionMsgOffset _winningOffset = null;
+ private String _winner;
+ private final PinotLLCRealtimeSegmentManager _segmentManager;
+ private final SegmentCompletionManager _segmentCompletionManager;
+ private final long _maxTimeToPickWinnerMs;
+ private final long _maxTimeToNotifyWinnerMs;
+ private final long _initialCommitTimeMs;
+ // Once the winner is notified, they are expected to commit right away. At this point, it is the segment build
+ // time that we need to consider.
+ // We may need to add some time here to allow for getting the lock? For now 0
+ // We may need to add some time for the committer come back to us (after the build)? For now 0.
+ private long _maxTimeAllowedToCommitMs;
+ private final String _controllerVipUrl;
+
+ public BlockingSegmentCompletionFSM(PinotLLCRealtimeSegmentManager segmentManager,
+ SegmentCompletionManager segmentCompletionManager, LLCSegmentName segmentName,
+ SegmentZKMetadata segmentMetadata) {
+ _segmentName = segmentName;
+ _rawTableName = _segmentName.getTableName();
+ _realtimeTableName = TableNameBuilder.REALTIME.tableNameWithType(_rawTableName);
+ _numReplicas = segmentMetadata.getNumReplicas();
+ _segmentManager = segmentManager;
+ _commitStateMap = new HashMap<>(HashUtil.getHashMapCapacity(_numReplicas));
+ _excludedServerStateMap = new HashSet<>(_numReplicas);
+ _segmentCompletionManager = segmentCompletionManager;
+ _startTimeMs = _segmentCompletionManager.getCurrentTimeMs();
+ _maxTimeToPickWinnerMs = _startTimeMs + MAX_TIME_TO_PICK_WINNER_MS;
+ _maxTimeToNotifyWinnerMs = _startTimeMs + MAX_TIME_TO_NOTIFY_WINNER_MS;
+ _streamPartitionMsgOffsetFactory = _segmentCompletionManager.getStreamPartitionMsgOffsetFactory(_segmentName);
+ long initialCommitTimeMs = MAX_TIME_TO_NOTIFY_WINNER_MS + _segmentManager.getCommitTimeoutMS(_realtimeTableName);
+ Long savedCommitTime = _segmentCompletionManager.getCommitTime(_rawTableName);
+ if (savedCommitTime != null && savedCommitTime > initialCommitTimeMs) {
+ initialCommitTimeMs = savedCommitTime;
+ }
+ _logger = LoggerFactory.getLogger("SegmentCompletionFSM_" + segmentName.getSegmentName());
+ int maxCommitTimeForAllSegmentsSeconds = SegmentCompletionManager.getMaxCommitTimeForAllSegmentsSeconds();
+ if (initialCommitTimeMs > maxCommitTimeForAllSegmentsSeconds * 1000L) {
+ // The table has a really high value configured for max commit time. Set it to a higher value than default
+ // and go from there.
+ _logger
+ .info("Configured max commit time {}s too high for table {}, changing to {}s", initialCommitTimeMs / 1000,
+ _realtimeTableName, maxCommitTimeForAllSegmentsSeconds);
+ initialCommitTimeMs = maxCommitTimeForAllSegmentsSeconds * 1000L;
+ }
+ _initialCommitTimeMs = initialCommitTimeMs;
+ _maxTimeAllowedToCommitMs = _startTimeMs + _initialCommitTimeMs;
+ _controllerVipUrl = segmentCompletionManager.getControllerVipUrl();
+
+ if (segmentMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.DONE) {
+ StreamPartitionMsgOffsetFactory factory =
+ _segmentCompletionManager.getStreamPartitionMsgOffsetFactory(_segmentName);
+ StreamPartitionMsgOffset endOffset = factory.create(segmentMetadata.getEndOffset());
+ _state = BlockingSegmentCompletionFSMState.COMMITTED;
+ _winningOffset = endOffset;
+ _winner = "UNKNOWN";
+ }
+ }
+
+ @Override
+ /**
+ * A new method that sets the initial FSM state based on the incoming message type.
+ */
Review Comment:
This javadoc seems redundant
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSMFactory.java:
##########
@@ -0,0 +1,132 @@
+/**
+ * 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.realtime;
+
+import com.google.common.base.Preconditions;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SegmentCompletionFSMFactory {
+ private SegmentCompletionFSMFactory() {
+ // Private constructor to prevent instantiation
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SegmentCompletionFSMFactory.class);
+ private static final Map<String, Class<? extends SegmentCompletionFSM>> FSM_CLASS_MAP = new HashMap<>();
+
+ // Static block to register the default FSM
+ static {
+ try {
+ Class<?> clazz = Class.forName(BlockingSegmentCompletionFSM.class.getCanonicalName());
+ register(SegmentCompletionConfig.DEFAULT_FSM_SCHEME, (Class<? extends SegmentCompletionFSM>) clazz);
+ LOGGER.info("Registered default BlockingSegmentCompletionFSM");
+ } catch (Exception e) {
+ LOGGER.error("Failed to register default BlockingSegmentCompletionFSM", e);
+ throw new RuntimeException("Failed to register default BlockingSegmentCompletionFSM", e);
+ }
Review Comment:
Can be simplified
```suggestion
register(SegmentCompletionConfig.DEFAULT_FSM_SCHEME, BlockingSegmentCompletionFSM.class);
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java:
##########
@@ -126,19 +127,14 @@ private SegmentCompletionFSM createFsm(LLCSegmentName llcSegmentName, String msg
SegmentZKMetadata segmentMetadata = _segmentManager.getSegmentZKMetadata(realtimeTableName, segmentName, null);
Preconditions.checkState(segmentMetadata != null, "Failed to find ZK metadata for segment: %s", segmentName);
SegmentCompletionFSM fsm;
- if (segmentMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.DONE) {
- // Segment is already committed
- StreamPartitionMsgOffsetFactory factory = getStreamPartitionMsgOffsetFactory(llcSegmentName);
- StreamPartitionMsgOffset endOffset = factory.create(segmentMetadata.getEndOffset());
- fsm = SegmentCompletionFSM.fsmInCommit(_segmentManager, this, llcSegmentName, segmentMetadata.getNumReplicas(),
- endOffset);
- } else if (msgType.equals(SegmentCompletionProtocol.MSG_TYPE_STOPPED_CONSUMING)) {
- fsm = SegmentCompletionFSM.fsmStoppedConsuming(_segmentManager, this, llcSegmentName,
- segmentMetadata.getNumReplicas());
- } else {
- // Segment is in the process of completing, and this is the first one to respond. Create fsm
- fsm = SegmentCompletionFSM.fsmInHolding(_segmentManager, this, llcSegmentName, segmentMetadata.getNumReplicas());
- }
+
+ String factoryName = _segmentCompletionConfig.getDefaultFsmScheme();
Review Comment:
Should we take scheme from table streaming config, and use default only if it is not configured?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSMFactory.java:
##########
@@ -0,0 +1,132 @@
+/**
+ * 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.realtime;
+
+import com.google.common.base.Preconditions;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SegmentCompletionFSMFactory {
+ private SegmentCompletionFSMFactory() {
+ // Private constructor to prevent instantiation
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SegmentCompletionFSMFactory.class);
+ private static final Map<String, Class<? extends SegmentCompletionFSM>> FSM_CLASS_MAP = new HashMap<>();
+
+ // Static block to register the default FSM
+ static {
+ try {
+ Class<?> clazz = Class.forName(BlockingSegmentCompletionFSM.class.getCanonicalName());
+ register(SegmentCompletionConfig.DEFAULT_FSM_SCHEME, (Class<? extends SegmentCompletionFSM>) clazz);
+ LOGGER.info("Registered default BlockingSegmentCompletionFSM");
+ } catch (Exception e) {
+ LOGGER.error("Failed to register default BlockingSegmentCompletionFSM", e);
+ throw new RuntimeException("Failed to register default BlockingSegmentCompletionFSM", e);
+ }
+ }
+
+ /**
+ * Registers an FSM class with a specific scheme/type.
+ *
+ * @param scheme The scheme or type key.
+ * @param fsmClass The class for the FSM.
+ */
+ public static void register(String scheme, Class<? extends SegmentCompletionFSM> fsmClass) {
+ Preconditions.checkNotNull(scheme, "Scheme cannot be null");
+ Preconditions.checkNotNull(fsmClass, "FSM Class cannot be null");
+ if (FSM_CLASS_MAP.containsKey(scheme)) {
+ LOGGER.warn("Overwriting existing FSM class for scheme {}", scheme);
+ }
+ FSM_CLASS_MAP.put(scheme, fsmClass);
+ LOGGER.info("Registered SegmentCompletionFSM class for scheme {}", scheme);
Review Comment:
Let's also log the class name
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionConfig.java:
##########
@@ -0,0 +1,44 @@
+/**
+ * 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.realtime;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.spi.env.PinotConfiguration;
+
+
+public class SegmentCompletionConfig {
+ private final Map<String, String> _fsmSchemes = new HashMap<>();
+
+ public SegmentCompletionConfig(PinotConfiguration configuration) {
Review Comment:
What I meant is to call `config.subset("pinot.controller.segment.completion")` to simplify the key handling within the config class. `"pinot.controller.segment.completion"` can be defined as a constant under `CommonConstants.Controller`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSM.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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.realtime;
+
+import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
+import org.apache.pinot.controller.helix.core.realtime.segment.CommittingSegmentDescriptor;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+
+
+/**
+ * Interface for managing the state machine transitions related to segment completion
+ * in a real-time table within Apache Pinot.
+ *
+ * The segment completion process is crucial for handling real-time ingestion, ensuring
+ * that segments are correctly built, committed, or discarded based on the server's state
+ * and the protocol defined for low-level consumer (LLC) tables. This interface abstracts
+ * the controller-side logic for responding to server events during the lifecycle of a
+ * segment's completion.
+ */
+public interface SegmentCompletionFSM {
+
+ /**
+ * Initializes the FSM to its initial state based on the message type.
+ * @param msgType
Review Comment:
Can you document what message type can be expected
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Make segment completion FSM pluggable [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #14088:
URL: https://github.com/apache/pinot/pull/14088#discussion_r1883550124
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java:
##########
@@ -126,19 +127,14 @@ private SegmentCompletionFSM createFsm(LLCSegmentName llcSegmentName, String msg
SegmentZKMetadata segmentMetadata = _segmentManager.getSegmentZKMetadata(realtimeTableName, segmentName, null);
Preconditions.checkState(segmentMetadata != null, "Failed to find ZK metadata for segment: %s", segmentName);
SegmentCompletionFSM fsm;
- if (segmentMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.DONE) {
- // Segment is already committed
- StreamPartitionMsgOffsetFactory factory = getStreamPartitionMsgOffsetFactory(llcSegmentName);
- StreamPartitionMsgOffset endOffset = factory.create(segmentMetadata.getEndOffset());
- fsm = SegmentCompletionFSM.fsmInCommit(_segmentManager, this, llcSegmentName, segmentMetadata.getNumReplicas(),
- endOffset);
- } else if (msgType.equals(SegmentCompletionProtocol.MSG_TYPE_STOPPED_CONSUMING)) {
- fsm = SegmentCompletionFSM.fsmStoppedConsuming(_segmentManager, this, llcSegmentName,
- segmentMetadata.getNumReplicas());
- } else {
- // Segment is in the process of completing, and this is the first one to respond. Create fsm
- fsm = SegmentCompletionFSM.fsmInHolding(_segmentManager, this, llcSegmentName, segmentMetadata.getNumReplicas());
- }
+
+ String factoryName = _segmentCompletionConfig.getDefaultFsmScheme();
Review Comment:
good idea, now taking this from stream config map
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Make segment completion FSM pluggable [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps merged PR #14088:
URL: https://github.com/apache/pinot/pull/14088
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [WIP] Pluggable fsm [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #14088:
URL: https://github.com/apache/pinot/pull/14088#discussion_r1857564000
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionConfig.java:
##########
@@ -0,0 +1,44 @@
+/**
+ * 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.realtime;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.spi.env.PinotConfiguration;
+
+
+public class SegmentCompletionConfig {
+ private final Map<String, String> _fsmSchemes = new HashMap<>();
+
+ public SegmentCompletionConfig(PinotConfiguration configuration) {
Review Comment:
Should we add some prefix to this config? Such as `pinot.controller.segment.completion`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSMState.java:
##########
@@ -0,0 +1,30 @@
+/**
+ * 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.realtime;
+
+public enum SegmentCompletionFSMState {
Review Comment:
If we want this to be pluggable, we might want to use string instead of enum, unless it can cover all scenarios
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSM.java:
##########
@@ -0,0 +1,42 @@
+/**
+ * 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.realtime;
+
+import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
+import org.apache.pinot.controller.helix.core.realtime.segment.CommittingSegmentDescriptor;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+
+
+public interface SegmentCompletionFSM {
Review Comment:
Please add some javadoc to this class and all methods
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSMFactory.java:
##########
@@ -0,0 +1,144 @@
+/**
+ * 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.realtime;
+
+import com.google.common.base.Preconditions;
+import java.lang.reflect.Constructor;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SegmentCompletionFSMFactory {
+ private SegmentCompletionFSMFactory() {
+ // Private constructor to prevent instantiation
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SegmentCompletionFSMFactory.class);
+ private static final Map<String, Constructor<? extends SegmentCompletionFSM>> FSM_CONSTRUCTOR_MAP = new HashMap<>();
+
+ // Static block to register the default FSM
+ static {
+ try {
+ Class<?> clazz = Class.forName("org.apache.pinot.controller.helix.core.realtime.BlockingSegmentCompletionFSM");
+ Constructor<?> constructor = clazz.getConstructor(
+ PinotLLCRealtimeSegmentManager.class,
+ SegmentCompletionManager.class,
+ LLCSegmentName.class,
+ SegmentZKMetadata.class,
+ String.class
+ );
+ register("BlockingSegmentCompletionFSMFactory", (Constructor<? extends SegmentCompletionFSM>) constructor);
Review Comment:
Consider registering class instead of constructor?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSMFactory.java:
##########
@@ -0,0 +1,144 @@
+/**
+ * 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.realtime;
+
+import com.google.common.base.Preconditions;
+import java.lang.reflect.Constructor;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SegmentCompletionFSMFactory {
+ private SegmentCompletionFSMFactory() {
+ // Private constructor to prevent instantiation
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SegmentCompletionFSMFactory.class);
+ private static final Map<String, Constructor<? extends SegmentCompletionFSM>> FSM_CONSTRUCTOR_MAP = new HashMap<>();
+
+ // Static block to register the default FSM
+ static {
+ try {
+ Class<?> clazz = Class.forName("org.apache.pinot.controller.helix.core.realtime.BlockingSegmentCompletionFSM");
+ Constructor<?> constructor = clazz.getConstructor(
+ PinotLLCRealtimeSegmentManager.class,
+ SegmentCompletionManager.class,
+ LLCSegmentName.class,
+ SegmentZKMetadata.class,
+ String.class
Review Comment:
Do we need a different FSM for each message type? Shouldn't different messages share the same FSM?
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [WIP] Pluggable fsm [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #14088:
URL: https://github.com/apache/pinot/pull/14088#discussion_r1861838450
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSMState.java:
##########
@@ -0,0 +1,30 @@
+/**
+ * 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.realtime;
+
+public enum SegmentCompletionFSMState {
Review Comment:
Not really, this is only used in the Blocking FSM implementation and not in interface or SegmentCompletionManager
other implementations can have their own enums
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [WIP] Pluggable fsm [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #14088:
URL: https://github.com/apache/pinot/pull/14088#discussion_r1862173178
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionFSMFactory.java:
##########
@@ -0,0 +1,144 @@
+/**
+ * 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.realtime;
+
+import com.google.common.base.Preconditions;
+import java.lang.reflect.Constructor;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SegmentCompletionFSMFactory {
+ private SegmentCompletionFSMFactory() {
+ // Private constructor to prevent instantiation
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SegmentCompletionFSMFactory.class);
+ private static final Map<String, Constructor<? extends SegmentCompletionFSM>> FSM_CONSTRUCTOR_MAP = new HashMap<>();
+
+ // Static block to register the default FSM
+ static {
+ try {
+ Class<?> clazz = Class.forName("org.apache.pinot.controller.helix.core.realtime.BlockingSegmentCompletionFSM");
+ Constructor<?> constructor = clazz.getConstructor(
+ PinotLLCRealtimeSegmentManager.class,
+ SegmentCompletionManager.class,
+ LLCSegmentName.class,
+ SegmentZKMetadata.class,
+ String.class
Review Comment:
No it's not different fsm but the issue is based on message type each fsm starts in a different state.
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org