This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 1c2a4fa13e6 Remove batch message mode and never allow enabling it
(#18795)
1c2a4fa13e6 is described below
commit 1c2a4fa13e6557862b893c7cfd7181ffe8e46891
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Wed Jun 17 16:52:07 2026 -0700
Remove batch message mode and never allow enabling it (#18795)
---
.../common/utils/helix/IdealStateGroupCommit.java | 7 +++-
.../common/utils/helix/IdealStateSingleCommit.java | 7 +++-
.../pinot/controller/BaseControllerStarter.java | 5 +--
.../apache/pinot/controller/ControllerConf.java | 6 ----
.../helix/core/PinotHelixResourceManager.java | 17 +++-------
.../helix/core/PinotTableIdealStateBuilder.java | 4 +--
.../helix/core/util/HelixSetupUtils.java | 38 ++++++++++------------
...otHelixResourceManagerConfigValidationTest.java | 3 +-
.../helix/core/retention/RetentionManagerTest.java | 6 ++--
9 files changed, 40 insertions(+), 53 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/IdealStateGroupCommit.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/IdealStateGroupCommit.java
index 4a2d83c65c2..b2710b7e70d 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/IdealStateGroupCommit.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/IdealStateGroupCommit.java
@@ -249,9 +249,14 @@ public class IdealStateGroupCommit {
// If there are changes to apply, apply them
if (updatedIdealState != null &&
!idealState.equals(updatedIdealState)) {
- ZNRecord updatedZNRecord = updatedIdealState.getRecord();
+ // Disable batch message mode if set
+ if (updatedIdealState.getBatchMessageMode()) {
+ LOGGER.warn("Disabling batch message mode for resource: {}",
resourceName);
+ updatedIdealState.setBatchMessageMode(false);
+ }
// Update number of partitions
+ ZNRecord updatedZNRecord = updatedIdealState.getRecord();
int numPartitions = updatedZNRecord.getMapFields().size();
updatedIdealState.setNumPartitions(numPartitions);
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/IdealStateSingleCommit.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/IdealStateSingleCommit.java
index a93237ee64d..c6b915d38ba 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/IdealStateSingleCommit.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/IdealStateSingleCommit.java
@@ -81,9 +81,14 @@ public class IdealStateSingleCommit {
// If there are changes to apply, apply them
if (updatedIdealState != null &&
!idealState.equals(updatedIdealState)) {
- ZNRecord updatedZNRecord = updatedIdealState.getRecord();
+ // Disable batch message mode if set
+ if (updatedIdealState.getBatchMessageMode()) {
+ LOGGER.warn("Disabling batch message mode for resource: {}",
resourceName);
+ updatedIdealState.setBatchMessageMode(false);
+ }
// Update number of partitions
+ ZNRecord updatedZNRecord = updatedIdealState.getRecord();
int numPartitions = updatedZNRecord.getMapFields().size();
updatedIdealState.setNumPartitions(numPartitions);
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
index a7ef8f42662..2d19445f486 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
@@ -203,7 +203,6 @@ public abstract class BaseControllerStarter implements
ServiceStartable {
protected String _helixControllerInstanceId;
protected String _helixParticipantInstanceId;
protected boolean _isUpdateStateModel;
- protected boolean _enableBatchMessageMode;
protected ControllerConf.ControllerMode _controllerMode;
protected HelixManager _helixControllerManager;
protected HelixManager _helixParticipantManager;
@@ -286,7 +285,6 @@ public abstract class BaseControllerStarter implements
ServiceStartable {
_helixParticipantInstanceId =
LeadControllerUtils.generateParticipantInstanceId(_hostname, _port);
}
_isUpdateStateModel = _config.isUpdateSegmentStateModel();
- _enableBatchMessageMode = _config.getEnableBatchMessageMode();
_serviceStatusCallbackList = new ArrayList<>();
if (_controllerMode == ControllerConf.ControllerMode.HELIX_ONLY) {
@@ -577,8 +575,7 @@ public abstract class BaseControllerStarter implements
ServiceStartable {
}
// Set up Pinot cluster in Helix if needed
- HelixSetupUtils.setupPinotCluster(_helixClusterName, _helixZkURL,
_isUpdateStateModel, _enableBatchMessageMode,
- _config);
+ HelixSetupUtils.setupPinotCluster(_helixClusterName, _helixZkURL,
_isUpdateStateModel, _config);
// Start all components
initPinotFSFactory();
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
index 9020f0be211..ead5eb03547 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
@@ -388,7 +388,6 @@ public class ControllerConf extends PinotConfiguration {
// the controller as ready. When enabled,
controller.resource.utilization.checker.initial.delay is set to 0.
public static final String
RESOURCE_UTILIZATION_CHECKER_COLLECT_USAGE_AT_STARTUP =
"controller.resource.utilization.checker.collect.usage.at.startup";
- public static final String ENABLE_BATCH_MESSAGE_MODE =
"controller.enable.batch.message.mode";
public static final String ENABLE_HYBRID_TABLE_RETENTION_STRATEGY =
"controller.enable.hybrid.table.retention.strategy";
// When true (default), segment deletion refuses to remove segments that
participate in a live segment lineage
@@ -428,7 +427,6 @@ public class ControllerConf extends PinotConfiguration {
public static final long DEFAULT_RESOURCE_UTILIZATION_CHECKER_INITIAL_DELAY
= 300L; // 5 minutes
public static final long DEFAULT_RESOURCE_UTILIZATION_CHECKER_FREQUENCY =
300L; // 5 minutes
public static final boolean
DEFAULT_RESOURCE_UTILIZATION_CHECKER_COLLECT_USAGE_AT_STARTUP = false;
- public static final boolean DEFAULT_ENABLE_BATCH_MESSAGE_MODE = false;
public static final boolean DEFAULT_ENABLE_HYBRID_TABLE_RETENTION_STRATEGY =
false;
public static final boolean DEFAULT_LINEAGE_EXCLUSIVE_DELETE_ENABLED = true;
public static final String DEFAULT_CONTROLLER_MODE =
ControllerMode.DUAL.name();
@@ -1306,10 +1304,6 @@ public class ControllerConf extends PinotConfiguration {
return getProperty(ENABLE_DISK_UTILIZATION_CHECKER,
DEFAULT_ENABLE_DISK_UTILIZATION_CHECKER);
}
- public boolean getEnableBatchMessageMode() {
- return getProperty(ENABLE_BATCH_MESSAGE_MODE,
DEFAULT_ENABLE_BATCH_MESSAGE_MODE);
- }
-
public boolean isHybridTableRetentionStrategyEnabled() {
return getProperty(ENABLE_HYBRID_TABLE_RETENTION_STRATEGY,
DEFAULT_ENABLE_HYBRID_TABLE_RETENTION_STRATEGY);
}
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
index fa7ecb8d2a1..35bc0cb301a 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
@@ -240,7 +240,6 @@ public class PinotHelixResourceManager {
private final String _helixClusterName;
private final String _dataDir;
private final boolean _isSingleTenantCluster;
- private final boolean _enableBatchMessageMode;
private final int _deletedSegmentsRetentionInDays;
private final boolean _enableTieredSegmentAssignment;
@Nullable
@@ -266,14 +265,12 @@ public class PinotHelixResourceManager {
private volatile ZkClient _zkClient;
private volatile MaterializedViewConsistencyManager
_materializedViewConsistencyManager;
- public PinotHelixResourceManager(String helixClusterName, @Nullable String
dataDir,
- boolean isSingleTenantCluster, boolean enableBatchMessageMode, int
deletedSegmentsRetentionInDays,
- boolean enableTieredSegmentAssignment, LineageManager lineageManager,
+ public PinotHelixResourceManager(String helixClusterName, @Nullable String
dataDir, boolean isSingleTenantCluster,
+ int deletedSegmentsRetentionInDays, boolean
enableTieredSegmentAssignment, LineageManager lineageManager,
@Nullable ControllerConf controllerConf) {
_helixClusterName = helixClusterName;
_dataDir = dataDir;
_isSingleTenantCluster = isSingleTenantCluster;
- _enableBatchMessageMode = enableBatchMessageMode;
_deletedSegmentsRetentionInDays = deletedSegmentsRetentionInDays;
_enableTieredSegmentAssignment = enableTieredSegmentAssignment;
_controllerConf = controllerConf;
@@ -295,8 +292,7 @@ public class PinotHelixResourceManager {
}
public PinotHelixResourceManager(ControllerConf controllerConf) {
- this(controllerConf.getHelixClusterName(), controllerConf.getDataDir(),
- controllerConf.tenantIsolationEnabled(),
controllerConf.getEnableBatchMessageMode(),
+ this(controllerConf.getHelixClusterName(), controllerConf.getDataDir(),
controllerConf.tenantIsolationEnabled(),
controllerConf.getDeletedSegmentsRetentionInDays(),
controllerConf.tieredSegmentAssignmentEnabled(),
LineageManagerFactory.create(controllerConf), controllerConf);
}
@@ -2097,8 +2093,7 @@ public class PinotHelixResourceManager {
LOGGER.info("Adding table {}: Successfully validated added table",
tableNameWithType);
IdealState idealState =
- PinotTableIdealStateBuilder.buildEmptyIdealStateFor(tableNameWithType,
tableConfig.getReplication(),
- _enableBatchMessageMode);
+ PinotTableIdealStateBuilder.buildEmptyIdealStateFor(tableNameWithType,
tableConfig.getReplication());
TableType tableType = tableConfig.getTableType();
// Ensure that table is not created if schema is not present
if (ZKMetadataProvider.getSchema(_propertyStore, rawTableName) == null) {
@@ -5480,15 +5475,13 @@ public class PinotHelixResourceManager {
final long externalViewOnlineToOfflineTimeoutMillis = 100L;
final boolean isSingleTenantCluster = false;
final boolean isUpdateStateModel = false;
- final boolean enableBatchMessageMode = false;
MetricsRegistry metricsRegistry = new MetricsRegistry();
final boolean dryRun = true;
final String tableName = "testTable";
final TableType tableType = TableType.OFFLINE;
PinotHelixResourceManager helixResourceManager =
new PinotHelixResourceManager(zkURL, helixClusterName,
controllerInstanceId, localDiskDir,
- externalViewOnlineToOfflineTimeoutMillis, isSingleTenantCluster,
isUpdateStateModel,
- * enableBatchMessageMode);
+ externalViewOnlineToOfflineTimeoutMillis, isSingleTenantCluster,
isUpdateStateModel);
helixResourceManager.start();
ZNRecord record = helixResourceManager.rebalanceTable(tableName, dryRun,
tableType);
ObjectMapper mapper = new ObjectMapper();
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotTableIdealStateBuilder.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotTableIdealStateBuilder.java
index fafbabb7ad3..8c2042aac67 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotTableIdealStateBuilder.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotTableIdealStateBuilder.java
@@ -42,15 +42,13 @@ public class PinotTableIdealStateBuilder {
private static final RetryPolicy DEFAULT_IDEALSTATE_UPDATE_RETRY_POLICY =
RetryPolicies.randomDelayRetryPolicy(3, 100L, 200L);
- public static IdealState buildEmptyIdealStateFor(String tableNameWithType,
int numReplicas,
- boolean enableBatchMessageMode) {
+ public static IdealState buildEmptyIdealStateFor(String tableNameWithType,
int numReplicas) {
CustomModeISBuilder customModeIdealStateBuilder = new
CustomModeISBuilder(tableNameWithType);
customModeIdealStateBuilder
.setStateModel(PinotHelixSegmentOnlineOfflineStateModelGenerator.PINOT_SEGMENT_ONLINE_OFFLINE_STATE_MODEL)
.setNumPartitions(0).setNumReplica(numReplicas).setMaxPartitionsPerNode(1);
IdealState idealState = customModeIdealStateBuilder.build();
idealState.setInstanceGroupTag(tableNameWithType);
- idealState.setBatchMessageMode(enableBatchMessageMode);
return idealState;
}
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/HelixSetupUtils.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/HelixSetupUtils.java
index 91d8ea163d2..470a7e320c6 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/HelixSetupUtils.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/HelixSetupUtils.java
@@ -133,7 +133,7 @@ public class HelixSetupUtils {
}
public static void setupPinotCluster(String helixClusterName, String zkPath,
boolean isUpdateStateModel,
- boolean enableBatchMessageMode, ControllerConf controllerConf) {
+ ControllerConf controllerConf) {
ZkClient zkClient = null;
int zkClientSessionConfig =
controllerConf.getProperty(CommonConstants.Helix.ZkClient.ZK_CLIENT_SESSION_TIMEOUT_MS_CONFIG,
@@ -161,11 +161,10 @@ public class HelixSetupUtils {
addSegmentStateModelDefinitionIfNeeded(helixClusterName, helixAdmin,
helixDataAccessor, isUpdateStateModel);
// Add broker resource if needed
- createBrokerResourceIfNeeded(helixClusterName, helixAdmin,
enableBatchMessageMode);
+ createBrokerResourceIfNeeded(helixClusterName, helixAdmin);
// Add lead controller resource if needed
- createLeadControllerResourceIfNeeded(helixClusterName, helixAdmin,
configAccessor, enableBatchMessageMode,
- controllerConf);
+ createLeadControllerResourceIfNeeded(helixClusterName, helixAdmin,
configAccessor, controllerConf);
} finally {
ZkStarter.closeAsync(zkClient);
}
@@ -187,8 +186,7 @@ public class HelixSetupUtils {
}
}
- private static void createBrokerResourceIfNeeded(String helixClusterName,
HelixAdmin helixAdmin,
- boolean enableBatchMessageMode) {
+ private static void createBrokerResourceIfNeeded(String helixClusterName,
HelixAdmin helixAdmin) {
// Add state model definition if needed
String stateModel =
PinotHelixBrokerResourceOnlineOfflineStateModelGenerator.PINOT_BROKER_RESOURCE_ONLINE_OFFLINE_STATE_MODEL;
@@ -200,24 +198,27 @@ public class HelixSetupUtils {
}
// Add broker resource if needed
- if (helixAdmin.getResourceIdealState(helixClusterName,
BROKER_RESOURCE_INSTANCE) == null) {
+ IdealState currentIdealState =
helixAdmin.getResourceIdealState(helixClusterName, BROKER_RESOURCE_INSTANCE);
+ if (currentIdealState == null) {
LOGGER.info("Adding resource: {}", BROKER_RESOURCE_INSTANCE);
IdealState idealState = new
CustomModeISBuilder(BROKER_RESOURCE_INSTANCE).setStateModel(stateModel).build();
- idealState.setBatchMessageMode(enableBatchMessageMode);
helixAdmin.addResource(helixClusterName, BROKER_RESOURCE_INSTANCE,
idealState);
+ } else if (currentIdealState.getBatchMessageMode()) {
+ LOGGER.warn("Disabling batch message mode for resource: {}",
BROKER_RESOURCE_INSTANCE);
+ currentIdealState.setBatchMessageMode(false);
+ helixAdmin.updateIdealState(helixClusterName, BROKER_RESOURCE_INSTANCE,
currentIdealState);
}
}
private static void createLeadControllerResourceIfNeeded(String
helixClusterName, HelixAdmin helixAdmin,
- ConfigAccessor configAccessor, boolean enableBatchMessageMode,
ControllerConf controllerConf) {
+ ConfigAccessor configAccessor, ControllerConf controllerConf) {
IdealState currentIdealState =
helixAdmin.getResourceIdealState(helixClusterName,
LEAD_CONTROLLER_RESOURCE_NAME);
if (currentIdealState == null) {
LOGGER.info("Adding resource: {}", LEAD_CONTROLLER_RESOURCE_NAME);
- IdealState newIdealState = constructIdealState(enableBatchMessageMode,
controllerConf);
+ IdealState newIdealState = constructIdealState(controllerConf);
helixAdmin.addResource(helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME,
newIdealState);
} else {
- enableAndUpdateLeadControllerResource(helixClusterName, helixAdmin,
currentIdealState, enableBatchMessageMode,
- controllerConf);
+ enableAndUpdateLeadControllerResource(helixClusterName, helixAdmin,
currentIdealState, controllerConf);
}
// Create resource config for lead controller resource if it doesn't exist
@@ -232,7 +233,7 @@ public class HelixSetupUtils {
configAccessor.setResourceConfig(helixClusterName,
LEAD_CONTROLLER_RESOURCE_NAME, resourceConfig);
}
- private static IdealState constructIdealState(boolean
enableBatchMessageMode, ControllerConf controllerConf) {
+ private static IdealState constructIdealState(ControllerConf controllerConf)
{
// FULL-AUTO Master-Slave state model with a rebalance strategy,
auto-rebalance by default
FullAutoModeISBuilder idealStateBuilder = new
FullAutoModeISBuilder(LEAD_CONTROLLER_RESOURCE_NAME);
idealStateBuilder.setStateModel(MasterSlaveSMD.name)
@@ -256,8 +257,6 @@ public class HelixSetupUtils {
// Set instance group tag
IdealState idealState = idealStateBuilder.build();
idealState.setInstanceGroupTag(CONTROLLER_INSTANCE);
- // Set batch message mode
- idealState.setBatchMessageMode(enableBatchMessageMode);
return idealState;
}
@@ -265,7 +264,7 @@ public class HelixSetupUtils {
* If user defined properties for the lead controller have changed, update
the resource.
*/
private static void enableAndUpdateLeadControllerResource(String
helixClusterName, HelixAdmin helixAdmin,
- IdealState idealState, boolean enableBatchMessageMode, ControllerConf
controllerConf) {
+ IdealState idealState, ControllerConf controllerConf) {
boolean needsUpdating = false;
if (!idealState.isEnabled()) {
@@ -275,10 +274,9 @@ public class HelixSetupUtils {
idealState.enable(true);
needsUpdating = true;
}
- if (idealState.getBatchMessageMode() != enableBatchMessageMode) {
- LOGGER.info("Updating batch message mode to: {} for resource: {}",
enableBatchMessageMode,
- LEAD_CONTROLLER_RESOURCE_NAME);
- idealState.setBatchMessageMode(enableBatchMessageMode);
+ if (idealState.getBatchMessageMode()) {
+ LOGGER.warn("Disabling batch message mode for resource: {}",
LEAD_CONTROLLER_RESOURCE_NAME);
+ idealState.setBatchMessageMode(false);
needsUpdating = true;
}
if
(!idealState.getRebalanceStrategy().equals(controllerConf.getLeadControllerResourceRebalanceStrategy()))
{
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerConfigValidationTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerConfigValidationTest.java
index a2b9f78d405..251bec9e955 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerConfigValidationTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerConfigValidationTest.java
@@ -52,8 +52,7 @@ public class PinotHelixResourceManagerConfigValidationTest {
public void setUp()
throws Exception {
LineageManager lineageManager = Mockito.mock(LineageManager.class);
- _resourceManager =
- new PinotHelixResourceManager("testCluster", null, false, false, 7,
false, lineageManager, null);
+ _resourceManager = new PinotHelixResourceManager("testCluster", null,
false, 7, false, lineageManager, null);
_helixAdmin = Mockito.mock(HelixAdmin.class);
_helixDataAccessor = Mockito.mock(HelixDataAccessor.class);
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
index 9999612bd24..042295c38ac 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
@@ -721,8 +721,7 @@ public class RetentionManagerTest {
List<SegmentZKMetadata> segmentsZKMetadata = new ArrayList<>();
- IdealState idealState =
-
PinotTableIdealStateBuilder.buildEmptyIdealStateFor(REALTIME_TABLE_NAME,
replicaCount, true);
+ IdealState idealState =
PinotTableIdealStateBuilder.buildEmptyIdealStateFor(REALTIME_TABLE_NAME,
replicaCount);
final int kafkaPartition = 5;
final long millisInDays = TimeUnit.DAYS.toMillis(1);
@@ -785,8 +784,7 @@ public class RetentionManagerTest {
List<SegmentZKMetadata> segmentsZKMetadata = new ArrayList<>();
- IdealState idealState =
-
PinotTableIdealStateBuilder.buildEmptyIdealStateFor(REALTIME_TABLE_NAME,
replicaCount, true);
+ IdealState idealState =
PinotTableIdealStateBuilder.buildEmptyIdealStateFor(REALTIME_TABLE_NAME,
replicaCount);
final int kafkaPartition = 5;
final long millisInDays = TimeUnit.DAYS.toMillis(1);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
