xiangfu0 commented on code in PR #20109: URL: https://github.com/apache/pinot/pull/20109#discussion_r4489378645
########## pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SharedOfflineUploadIntegrationTestSuite.java: ########## @@ -0,0 +1,307 @@ +/** + * 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.integration.tests; + +import java.io.File; +import java.io.IOException; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; +import org.apache.commons.io.FileUtils; +import org.apache.pinot.client.admin.PinotAdminClient; +import org.apache.pinot.integration.tests.custom.CustomDataQueryClusterIntegrationTest; +import org.apache.pinot.server.starter.helix.BaseServerStarter; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.utils.builder.TableNameBuilder; +import org.apache.pinot.util.TestUtils; +import org.testng.ITestResult; +import org.testng.annotations.AfterClass; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.AfterSuite; +import org.testng.annotations.BeforeClass; + + +/// Runs segment upload scenarios against the custom suite cluster. +/// Each class owns its table and schema resources; parallel classes are rejected. +public abstract class SharedOfflineUploadIntegrationTestSuite extends CustomDataQueryClusterIntegrationTest { + private static final Set<SharedOfflineUploadIntegrationTestSuite> PARTICIPANTS = new LinkedHashSet<>(); + private static SharedOfflineUploadIntegrationTestSuite _activeClass; + + private final Set<String> _tables = new LinkedHashSet<>(); + private final Set<String> _schemas = new LinkedHashSet<>(); + + @BeforeClass(alwaysRun = true) + public final void bindSharedOfflineUploadCluster() { + synchronized (SharedOfflineUploadIntegrationTestSuite.class) { + assertCustomSuiteHealthy(); + if (_activeClass != null) { + throw new IllegalStateException("Another offline upload class still owns the shared cluster"); + } + _activeClass = this; + PARTICIPANTS.add(this); + // These inherited fields are read directly by metadata-push tests and external-view helpers. + _controllerConfig = getSharedControllerStarter().getConfig(); + _controllerBaseApiUrl = _controllerConfig.generateVipUrl(); + _helixResourceManager = getSharedHelixResourceManager(); + initControllerRequestURLBuilder(); + } + } + + // These tests create their own empty table at the point required by each scenario. + @Override + @BeforeClass + public void setUp() + throws Exception { + } + + @Override + @AfterClass(alwaysRun = true) + public void tearDown() + throws IOException { + } + + @Override + public String getTableName() { + return DEFAULT_TABLE_NAME; + } + + @Override + public Schema createSchema() + throws IOException { + return createDefaultSchema(); + } + + @Override + public List<File> createAvroFiles() + throws Exception { + return getAllAvroFiles(); + } + + @Override + public TableConfig createOfflineTableConfig() { + return createDefaultOfflineTableConfig(); + } + + @Override + public String getTimeColumnName() { + return DEFAULT_TIME_COLUMN_NAME; + } + + @Override + protected String getSortedColumn() { + return getDefaultSortedColumn(); + } + + @Override + protected List<String> getInvertedIndexColumns() { + return getDefaultInvertedIndexColumns(); + } + + @Override + protected List<String> getNoDictionaryColumns() { + return getDefaultNoDictionaryColumns(); + } + + @Override + protected List<String> getRangeIndexColumns() { + return getDefaultRangeIndexColumns(); + } + + @Override + protected List<String> getBloomFilterColumns() { + return getDefaultBloomFilterColumns(); + } + + @Override + public final void addSchema(Schema schema) + throws IOException { + _schemas.add(schema.getSchemaName()); + super.addSchema(schema); + } + + @Override + public final void addTableConfig(TableConfig tableConfig) + throws IOException { + if (tableConfig.getTableType() != TableType.OFFLINE) { + throw new IllegalArgumentException("The shared offline upload suite only supports OFFLINE tables"); + } + _tables.add(tableConfig.getTableName()); + super.addTableConfig(tableConfig); + } + + @AfterMethod(alwaysRun = true) + public final void cleanUpSharedOfflineUploadMethod(ITestResult result) + throws Throwable { + Throwable failure = cleanUpTables(null); + if (failure != null) { + recordCleanupFailure(failure); + if (result.getThrowable() != null && result.getThrowable() != failure) { + result.getThrowable().addSuppressed(failure); + } + throw failure; + } + } + + @AfterClass(alwaysRun = true) + public final void cleanUpSharedOfflineUploadClass() + throws Throwable { + Throwable failure = cleanUpClass(null); + synchronized (SharedOfflineUploadIntegrationTestSuite.class) { + if (_activeClass == this) { + _activeClass = null; + } + } + if (failure != null) { + recordCleanupFailure(failure); + throw failure; + } + } + + private Throwable cleanUpTables(Throwable failure) { + for (String table : List.copyOf(_tables)) { + try { + removeTable(table); + _tables.remove(table); + } catch (Throwable t) { + failure = appendFailure(failure, t); + } + } + return failure; + } + + private Throwable cleanUpClass(Throwable failure) { + failure = cleanUpTables(failure); + // Preserve the original schema lifetime across upload methods, then isolate the next class. + if (_tables.isEmpty()) { + for (String schema : List.copyOf(_schemas)) { + try { + if (getSharedHelixResourceManager().getSchema(schema) != null) { + deleteSchema(schema); + } + _schemas.remove(schema); + } catch (Throwable t) { + failure = appendFailure(failure, t); + } + } + } + if (_pinotConnection != null) { + failure = cleanUp(failure, () -> { + _pinotConnection.close(); + _pinotConnection = null; + }); + } + if (_pinotConnectionV2 != null) { + failure = cleanUp(failure, () -> { + _pinotConnectionV2.close(); + _pinotConnectionV2 = null; + }); + } + if (_tables.isEmpty() && _schemas.isEmpty() && this != _sharedClusterTestSuite) { + // The suite owner retains its directory until the Custom suite teardown. + failure = cleanUp(failure, () -> FileUtils.deleteDirectory(_tempDir)); + } + return failure; + } + + @AfterSuite(alwaysRun = true) + public final void tearDownSharedOfflineUploadSuite() + throws Throwable { + Throwable failure = null; + for (SharedOfflineUploadIntegrationTestSuite participant : PARTICIPANTS) { + failure = participant.cleanUpClass(failure); + } + PARTICIPANTS.clear(); + _activeClass = null; + if (failure != null) { + recordCleanupFailure(failure); + throw failure; + } + } + + private static void recordCleanupFailure(Throwable failure) { + recordCustomSuiteCleanupFailure(failure); + } + + private static Throwable appendFailure(Throwable failure, Throwable next) { + if (failure == null) { + return next; + } + if (failure != next) { + failure.addSuppressed(next); + } + return failure; + } + + private static Throwable cleanUp(Throwable failure, CleanupAction action) { + try { + action.run(); + } catch (Throwable t) { + return appendFailure(failure, t); + } + return failure; + } + + /// A cleanup step whose failure must not prevent the remaining steps. + @FunctionalInterface + private interface CleanupAction { + void run() + throws Exception; + } + + @Override + public final PinotAdminClient getOrCreateAdminClient() + throws IOException { + if (_sharedClusterTestSuite == this) { + return super.getOrCreateAdminClient(); + } + return _sharedClusterTestSuite.getOrCreateAdminClient(); + } + + private void removeTable(String tableNameWithType) + throws IOException { + List<BaseServerStarter> servers = getSharedServerStarters(); + if (servers.size() != 2) { Review Comment: Hard-coding `2` couples cleanup to `CustomDataQueryClusterIntegrationTest#startServer`. If that server count ever changes, every upload cleanup throws, and the health guard then skips the rest of the Custom suite. Checking `servers.isEmpty()`, or iterating over whatever servers are present, keeps the intent without the coupling. ########## pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomSuiteHealthCheckedIntegrationTest.java: ########## @@ -0,0 +1,72 @@ +/** + * 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.integration.tests.custom; + +import org.apache.pinot.integration.tests.BaseClusterIntegrationTest; +import org.testng.annotations.AfterSuite; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.BeforeSuite; + + +/// Fails closed after a shared custom-cluster participant reports a cleanup failure. +/// Supports one sequential suite per JVM; concurrent suites in the same JVM are not supported. +/// Synchronized access protects failure reporting between callbacks. +public abstract class CustomSuiteHealthCheckedIntegrationTest extends BaseClusterIntegrationTest { + private static Throwable _customSuiteCleanupFailure; + + @BeforeSuite(alwaysRun = true) + public final void resetCustomSuiteHealthBeforeSuite() { + resetCustomSuiteHealth(); + } + + @AfterSuite(alwaysRun = true) + public final void resetCustomSuiteHealthAfterSuite() { + resetCustomSuiteHealth(); + } + + @BeforeClass(alwaysRun = true) + public final void checkCustomSuiteHealthBeforeClass() { Review Comment: This needs a deliberate decision. One transient cleanup timeout in an upload class (for example, the 60s routing-404 wait) now skips every later class in the 865-test Custom suite, instead of failing just one class. The build fails either way, but that run loses most of its signal. If this is intended, fine; otherwise consider scoping the guard to the upload participants. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
