This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch codex/share-offline-upload-test-suite in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 6f88ec603eff965c9c0ab42a2b1c65dc1bea3dbc Author: Chen Liang <[email protected]> AuthorDate: Sun Nov 1 05:48:59 2026 -0800 Share custom cluster for offline upload integration tests --- .../common/compression/CompressionFactoryTest.java | 43 ++++ .../common/config/provider/ZkTableCacheTest.java | 91 +++++++ .../queries/DistinctCountBitmapRawQueriesTest.java | 182 ++++++++++++++ .../tests/BaseClusterIntegrationTest.java | 7 +- pinot-integration-tests/pom.xml | 19 -- .../tests/CustomSuiteHealthCheckTest.java | 69 +++++ .../integration/tests/CustomSuiteHealthProbe.java | 165 ++++++++++++ ...ite.java => CustomSuiteLifecycleCheckTest.java} | 39 ++- .../tests/CustomSuiteLifecycleProbe.java | 269 ++++++++++++++++++++ .../tests/SegmentUploadIntegrationTest.java | 1 + .../SegmentWriterUploaderIntegrationTest.java | 1 + .../SharedOfflineUploadIntegrationTestSuite.java | 280 +++++++++------------ .../CustomDataQueryClusterIntegrationTest.java | 242 ++++++++++++++++-- .../CustomSuiteHealthCheckedIntegrationTest.java | 72 ++++++ .../tests/suites/CustomClusterSuite.java | 4 + .../tests/suites/SharedOfflineUploadSuite.java | 41 --- 16 files changed, 1261 insertions(+), 264 deletions(-) diff --git a/pinot-common/src/test/java/org/apache/pinot/common/compression/CompressionFactoryTest.java b/pinot-common/src/test/java/org/apache/pinot/common/compression/CompressionFactoryTest.java new file mode 100644 index 00000000000..608f687669d --- /dev/null +++ b/pinot-common/src/test/java/org/apache/pinot/common/compression/CompressionFactoryTest.java @@ -0,0 +1,43 @@ +/** + * 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.common.compression; + +import java.nio.charset.StandardCharsets; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertSame; + + +/// Tests the case-insensitive names accepted by the compression factory. +public class CompressionFactoryTest { + @Test + public void testZstandardAliasRoundTrip() + throws Exception { + Compressor zstd = CompressionFactory.getCompressor("ZSTD"); + Compressor alias = CompressionFactory.getCompressor("ZSTANDARD"); + Compressor mixedCaseAlias = CompressionFactory.getCompressor("zStAnDaRd"); + + assertSame(alias, zstd); + assertSame(mixedCaseAlias, zstd); + byte[] input = "Pinot segment compression alias test".getBytes(StandardCharsets.UTF_8); + assertEquals(zstd.decompress(alias.compress(input)), input); + assertEquals(mixedCaseAlias.decompress(zstd.compress(input)), input); + } +} diff --git a/pinot-common/src/test/java/org/apache/pinot/common/config/provider/ZkTableCacheTest.java b/pinot-common/src/test/java/org/apache/pinot/common/config/provider/ZkTableCacheTest.java new file mode 100644 index 00000000000..f4de36ec380 --- /dev/null +++ b/pinot-common/src/test/java/org/apache/pinot/common/config/provider/ZkTableCacheTest.java @@ -0,0 +1,91 @@ +/** + * 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.common.config.provider; + +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Collectors; +import org.apache.helix.AccessOption; +import org.apache.helix.store.zk.ZkHelixPropertyStore; +import org.apache.helix.zookeeper.datamodel.ZNRecord; +import org.apache.helix.zookeeper.zkclient.IZkChildListener; +import org.apache.helix.zookeeper.zkclient.IZkDataListener; +import org.apache.pinot.common.utils.config.SchemaSerDeUtils; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.data.Schema; +import org.mockito.ArgumentCaptor; +import org.testng.annotations.Test; + +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.ArgumentMatchers.same; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertTrue; + + +/// Tests schema child and data notifications in the ZooKeeper-backed table cache. +public class ZkTableCacheTest { + @Test + @SuppressWarnings("deprecation") // The cache currently calls the deprecated bulk-get overload. + public void testEmptySchemaChildrenAndLastSchemaDeletion() + throws Exception { + String schemaName = "testSchema"; + String schemaPath = "/SCHEMAS/" + schemaName; + Schema schema = new Schema.SchemaBuilder().setSchemaName(schemaName) + .addSingleValueDimension("column", DataType.INT).build(); + + @SuppressWarnings("unchecked") + ZkHelixPropertyStore<ZNRecord> propertyStore = mock(ZkHelixPropertyStore.class); + when(propertyStore.getChildNames("/SCHEMAS", AccessOption.PERSISTENT)).thenReturn(List.of(schemaName)); + when(propertyStore.get(eq(List.of(schemaPath)), isNull(), eq(AccessOption.PERSISTENT))) + .thenReturn(List.of(SchemaSerDeUtils.toZNRecord(schema))); + + ZkTableCache tableCache = new ZkTableCache(propertyStore, false); + ArgumentCaptor<IZkChildListener> childListener = ArgumentCaptor.forClass(IZkChildListener.class); + verify(propertyStore).subscribeChildChanges(eq("/SCHEMAS"), childListener.capture()); + IZkChildListener schemaChildListener = childListener.getValue(); + verify(propertyStore).subscribeDataChanges(eq(schemaPath), same((IZkDataListener) schemaChildListener)); + List<List<String>> notifications = new ArrayList<>(); + assertTrue(tableCache.registerSchemaChangeListener(schemas -> + notifications.add(schemas.stream().map(Schema::getSchemaName).collect(Collectors.toList())))); + assertNotNull(tableCache.getSchema(schemaName)); + assertEquals(notifications, List.of(List.of(schemaName))); + + // A child-list event may precede the data-deleted event. It must not remove the schema prematurely. + schemaChildListener.handleChildChange("/SCHEMAS", List.of()); + schemaChildListener.handleChildChange("/SCHEMAS", null); + assertNotNull(tableCache.getSchema(schemaName)); + assertEquals(notifications, List.of(List.of(schemaName))); + verify(propertyStore, times(1)).get(eq(List.of(schemaPath)), isNull(), eq(AccessOption.PERSISTENT)); + + ((IZkDataListener) schemaChildListener).handleDataDeleted("/cluster/PROPERTYSTORE" + schemaPath); + assertNull(tableCache.getSchema(schemaName)); + assertEquals(notifications, List.of(List.of(schemaName), List.of())); + + // The child-list event can also arrive after deletion without sending a duplicate notification. + schemaChildListener.handleChildChange("/SCHEMAS", List.of()); + assertEquals(notifications, List.of(List.of(schemaName), List.of())); + } +} diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/DistinctCountBitmapRawQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/DistinctCountBitmapRawQueriesTest.java new file mode 100644 index 00000000000..6796a0e6b7b --- /dev/null +++ b/pinot-core/src/test/java/org/apache/pinot/queries/DistinctCountBitmapRawQueriesTest.java @@ -0,0 +1,182 @@ +/** + * 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.queries; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.List; +import org.apache.commons.io.FileUtils; +import org.apache.pinot.common.response.broker.BrokerResponseNative; +import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader; +import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl; +import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader; +import org.apache.pinot.segment.spi.IndexSegment; +import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.utils.ReadMode; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNull; + + +/// Tests bitmap distinct aggregation on raw and dictionary-encoded values across two segments. +/// The test instance owns its segments and is not thread-safe. +public class DistinctCountBitmapRawQueriesTest extends BaseQueriesTest { + private static final String TABLE_NAME = "bitmapRawTestTable"; + private static final String GROUP_COLUMN = "groupCol"; + private static final String RAW_INT_COLUMN = "rawInt"; + private static final String RAW_DOUBLE_COLUMN = "rawDouble"; + private static final String DICT_DOUBLE_COLUMN = "dictDouble"; + private static final Schema SCHEMA = new Schema.SchemaBuilder() + .addSingleValueDimension(GROUP_COLUMN, DataType.STRING) + .addSingleValueDimension(RAW_INT_COLUMN, DataType.INT) + .addSingleValueDimension(RAW_DOUBLE_COLUMN, DataType.DOUBLE) + .addSingleValueDimension(DICT_DOUBLE_COLUMN, DataType.DOUBLE) + .build(); + private static final TableConfig TABLE_CONFIG = new TableConfigBuilder(TableType.OFFLINE) + .setTableName(TABLE_NAME) + .setNoDictionaryColumns(List.of(RAW_INT_COLUMN, RAW_DOUBLE_COLUMN)) + .build(); + + private final List<IndexSegment> _segments = new ArrayList<>(); + private File _indexDir; + + @Override + protected String getFilter() { + return ""; + } + + @Override + protected IndexSegment getIndexSegment() { + return _segments.get(0); + } + + @Override + protected List<IndexSegment> getIndexSegments() { + return _segments; + } + + @Override + protected List<List<IndexSegment>> getDistinctInstances() { + return List.of(List.of(_segments.get(0)), List.of(_segments.get(1))); + } + + @BeforeClass + public void setUp() + throws Exception { + _indexDir = Files.createTempDirectory("pinot-bitmap-raw-").toFile(); + _segments.add(buildSegment("segmentA", List.of( + row("a", 10, 1.0, 10.0), row("a", 10, 2.0, 10.0), row("a", 20, 3.0, 20.0), row("b", 30, 3.0, 30.0)))); + _segments.add(buildSegment("segmentB", List.of( + row("a", 20, 3.0, 20.0), row("a", 30, 4.0, 20.0), row("b", 30, 3.0, 30.0), row("b", 40, 3.0, 30.0)))); + + for (IndexSegment segment : _segments) { + assertNull(segment.getDataSource(RAW_INT_COLUMN).getDictionary()); + assertNull(segment.getDataSource(RAW_DOUBLE_COLUMN).getDictionary()); + assertNotNull(segment.getDataSource(DICT_DOUBLE_COLUMN).getDictionary()); + } + // Dict id 0 means a different value in each segment. A merge must use values, not bare dictionary ids. + assertEquals(_segments.get(0).getDataSource(DICT_DOUBLE_COLUMN).getDictionary().getDoubleValue(0), 10.0d); + assertEquals(_segments.get(1).getDataSource(DICT_DOUBLE_COLUMN).getDictionary().getDoubleValue(0), 20.0d); + } + + private static GenericRow row(String group, int rawInt, double rawDouble, double dictDouble) { + GenericRow record = new GenericRow(); + record.putValue(GROUP_COLUMN, group); + record.putValue(RAW_INT_COLUMN, rawInt); + record.putValue(RAW_DOUBLE_COLUMN, rawDouble); + record.putValue(DICT_DOUBLE_COLUMN, dictDouble); + return record; + } + + private IndexSegment buildSegment(String segmentName, List<GenericRow> records) + throws Exception { + SegmentGeneratorConfig config = new SegmentGeneratorConfig(TABLE_CONFIG, SCHEMA); + config.setTableName(TABLE_NAME); + config.setSegmentName(segmentName); + config.setOutDir(_indexDir.getPath()); + SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl(); + driver.init(config, new GenericRowRecordReader(records)); + driver.build(); + return ImmutableSegmentLoader.load(new File(_indexDir, segmentName), ReadMode.mmap); + } + + @Test + public void testGroupByMergesRawAndDictionaryValuesAcrossSegments() { + String query = "SELECT groupCol, DISTINCTCOUNTBITMAP(rawInt), DISTINCTCOUNTBITMAP(rawDouble), " + + "DISTINCTCOUNTBITMAP(dictDouble) FROM " + TABLE_NAME + " GROUP BY groupCol ORDER BY groupCol"; + BrokerResponseNative response = getBrokerResponse(query); + assertEquals(response.getExceptions().size(), 0, response.getExceptions().toString()); + assertNotNull(response.getResultTable()); + List<Object[]> rows = response.getResultTable().getRows(); + assertEquals(rows.size(), 2); + assertGroupRow(rows.get(0), "a", 3, 4, 2); + assertGroupRow(rows.get(1), "b", 2, 1, 1); + } + + private static void assertGroupRow(Object[] row, String group, int intCount, int rawDoubleCount, + int dictDoubleCount) { + assertEquals(row.length, 4); + assertEquals(row[0], group); + assertEquals(((Number) row[1]).intValue(), intCount); + assertEquals(((Number) row[2]).intValue(), rawDoubleCount); + assertEquals(((Number) row[3]).intValue(), dictDoubleCount); + } + + @Test + public void testAggregationMergesRawAndDictionaryValuesAcrossSegments() { + String query = "SELECT DISTINCTCOUNTBITMAP(rawInt), DISTINCTCOUNTBITMAP(rawDouble), " + + "DISTINCTCOUNTBITMAP(dictDouble) FROM " + TABLE_NAME; + BrokerResponseNative response = getBrokerResponse(query); + assertEquals(response.getExceptions().size(), 0, response.getExceptions().toString()); + assertNotNull(response.getResultTable()); + List<Object[]> rows = response.getResultTable().getRows(); + assertEquals(rows.size(), 1); + Object[] row = rows.get(0); + assertEquals(row.length, 3); + assertEquals(((Number) row[0]).intValue(), 4); + assertEquals(((Number) row[1]).intValue(), 4); + assertEquals(((Number) row[2]).intValue(), 3); + } + + @AfterClass(alwaysRun = true) + public void tearDown() + throws IOException { + try { + for (IndexSegment segment : _segments) { + segment.destroy(); + } + } finally { + if (_indexDir != null) { + FileUtils.deleteDirectory(_indexDir); + } + } + } +} diff --git a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java index e3dfbaf19a0..ac038fafc68 100644 --- a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java +++ b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java @@ -773,12 +773,17 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest { createKafkaTopic(getKafkaTopic()); } + /// Creates the embedded Kafka fixture; subclasses can retain the allocated instance during startup. + protected EmbeddedKafkaCluster createEmbeddedKafkaCluster() { + return new EmbeddedKafkaCluster(); + } + protected void startKafkaWithoutTopic() { int requestedBrokers = getNumKafkaBrokers(); Properties props = new Properties(); props.setProperty(EmbeddedKafkaCluster.BROKER_COUNT_PROP, Integer.toString(requestedBrokers)); props.putAll(getKafkaExtraProperties()); - EmbeddedKafkaCluster cluster = new EmbeddedKafkaCluster(); + EmbeddedKafkaCluster cluster = createEmbeddedKafkaCluster(); cluster.init(props); cluster.start(); _kafkaStarters = List.of(cluster); diff --git a/pinot-integration-tests/pom.xml b/pinot-integration-tests/pom.xml index 69b0440a17f..7db8c541d41 100644 --- a/pinot-integration-tests/pom.xml +++ b/pinot-integration-tests/pom.xml @@ -364,25 +364,6 @@ <exclude>org/apache/pinot/integration/tests/TenantRebalanceIntegrationTest.java</exclude> </excludes> </configuration> - <executions> - <execution> - <id>shared-offline-upload-integration-test-suite</id> - <phase>test</phase> - <goals> - <goal>test</goal> - </goals> - <configuration> - <skipTests>${pinot.integration.test.skip.named.suites}</skipTests> - <reportNameSuffix>shared-offline-upload</reportNameSuffix> - <excludes combine.self="override"> - <exclude>none</exclude> - </excludes> - <includes combine.self="override"> - <include>org/apache/pinot/integration/tests/suites/SharedOfflineUploadSuite.java</include> - </includes> - </configuration> - </execution> - </executions> </plugin> </plugins> </build> diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CustomSuiteHealthCheckTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CustomSuiteHealthCheckTest.java new file mode 100644 index 00000000000..f4235523c36 --- /dev/null +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CustomSuiteHealthCheckTest.java @@ -0,0 +1,69 @@ +/** + * 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.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.TimeUnit; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + + +/// Verifies the custom suite stops after a shared fixture cleanup failure. +public class CustomSuiteHealthCheckTest { + @Test + public void testCleanupFailureWithContinuePolicy() + throws Exception { + runInSeparateJvm("continue"); + } + + @Test + public void testCleanupFailureWithDefaultPolicy() + throws Exception { + runInSeparateJvm("default"); + } + + /// Runs nested TestNG outside the outer TestNG engine, so nested listeners cannot corrupt its state. + private static void runInSeparateJvm(String policy) + throws Exception { + String java = Path.of(System.getProperty("java.home"), "bin", "java").toString(); + String classPath = System.getProperty("surefire.test.class.path", System.getProperty("java.class.path")); + Path output = Files.createTempFile("custom-suite-health-", ".log"); + Process process = null; + try { + process = new ProcessBuilder(java, "-cp", classPath, CustomSuiteHealthProbe.class.getName(), policy) + .redirectErrorStream(true).redirectOutput(output.toFile()).start(); + boolean completed = process.waitFor(120, TimeUnit.SECONDS); + if (!completed) { + process.destroyForcibly(); + process.waitFor(10, TimeUnit.SECONDS); + } + String childOutput = Files.readString(output); + assertTrue(completed, "Health-check subprocess timed out: " + childOutput); + assertEquals(process.exitValue(), 0, "Health-check subprocess failed: " + childOutput); + } finally { + if (process != null && process.isAlive()) { + process.destroyForcibly(); + } + Files.deleteIfExists(output); + } + } +} diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CustomSuiteHealthProbe.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CustomSuiteHealthProbe.java new file mode 100644 index 00000000000..2541a7e0dc1 --- /dev/null +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CustomSuiteHealthProbe.java @@ -0,0 +1,165 @@ +/** + * 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.util.Arrays; +import java.util.List; +import org.apache.pinot.integration.tests.custom.CustomSuiteHealthCheckedIntegrationTest; +import org.testng.ITestResult; +import org.testng.TestListenerAdapter; +import org.testng.TestNG; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.AfterSuite; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; +import org.testng.xml.XmlClass; +import org.testng.xml.XmlSuite; +import org.testng.xml.XmlTest; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; + + +/// Subprocess-only TestNG fixture used to test custom-suite health handling. +public final class CustomSuiteHealthProbe { + private CustomSuiteHealthProbe() { + } + + private static RuntimeException _cleanupFailure; + private static boolean _laterMethodRan; + private static boolean _laterClassRan; + private static boolean _laterClassSetupRan; + private static boolean _suiteTearDownRan; + private static boolean _failureVisibleDuringTearDown; + + public static void main(String[] args) { + assertEquals(args.length, 1, "Expected a failure policy"); + switch (args[0]) { + case "continue": + verifyCleanupFailure(XmlSuite.FailurePolicy.CONTINUE); + break; + case "default": + verifyCleanupFailure(null); + break; + default: + throw new IllegalArgumentException("Unexpected failure policy: " + args[0]); + } + } + + private static void verifyCleanupFailure(XmlSuite.FailurePolicy policy) { + _cleanupFailure = new RuntimeException("injected cleanup failure"); + _laterMethodRan = false; + _laterClassRan = false; + _laterClassSetupRan = false; + _suiteTearDownRan = false; + _failureVisibleDuringTearDown = false; + XmlSuite suite = suite("injected-cleanup-failure", FailingParticipant.class, LaterParticipant.class); + if (policy != null) { + suite.setConfigFailurePolicy(policy); + } + TestListenerAdapter listener = run(suite); + + assertTrue(listener.getConfigurationFailures().stream() + .anyMatch(result -> result.getThrowable() == _cleanupFailure), "Original teardown failure must be reported"); + assertTrue(listener.getConfigurationFailures().stream() + .anyMatch(result -> result.getThrowable().getCause() == _cleanupFailure), + "A health check should retain the cleanup failure as its cause"); + assertTrue(listener.getSkippedTests().stream().anyMatch(result -> result.getName().equals("laterMethod"))); + assertTrue(listener.getSkippedTests().stream().anyMatch(result -> result.getName().equals("laterClass"))); + assertFalse(_laterMethodRan); + assertFalse(_laterClassRan); + assertFalse(_laterClassSetupRan, "The next class must not mutate the contaminated cluster during setup"); + assertTrue(_suiteTearDownRan, "Suite teardown must run even after cleanup fails"); + assertTrue(_failureVisibleDuringTearDown, "The guard must remain active until suite teardown"); + FailingParticipant.assertHealthyAfterSuite(); + + // A later, independent suite in the same JVM must not inherit the previous failure. + TestListenerAdapter next = run(suite("independent-suite", LaterParticipant.class)); + assertTrue(next.getConfigurationFailures().isEmpty()); + assertTrue(next.getFailedTests().isEmpty()); + assertTrue(_laterClassRan, "The next suite must start with a healthy guard"); + } + + private static XmlSuite suite(String name, Class<?>... classes) { + XmlSuite suite = new XmlSuite(); + suite.setName(name); + suite.setPreserveOrder(true); + XmlTest test = new XmlTest(suite); + test.setName(name); + test.setXmlClasses(Arrays.stream(classes).map(XmlClass::new).toList()); + return suite; + } + + private static TestListenerAdapter run(XmlSuite suite) { + TestListenerAdapter listener = new TestListenerAdapter(); + TestNG testng = new TestNG(false); + testng.setUseDefaultListeners(false); + testng.setVerbose(0); + testng.setXmlSuites(List.of(suite)); + testng.addListener(listener); + testng.run(); + return listener; + } + + public static class FailingParticipant extends CustomSuiteHealthCheckedIntegrationTest { + @Test + public void firstMethod() { + } + + @AfterMethod(alwaysRun = true) + public void failCleanup(ITestResult result) { + if (result.getMethod().getMethodName().equals("firstMethod")) { + recordCustomSuiteCleanupFailure(_cleanupFailure); + throw _cleanupFailure; + } + } + + @Test(dependsOnMethods = "firstMethod", alwaysRun = true) + public void laterMethod() { + _laterMethodRan = true; + } + + @AfterSuite(alwaysRun = true) + public void finishSuite() { + _suiteTearDownRan = true; + try { + assertCustomSuiteHealthy(); + } catch (IllegalStateException e) { + _failureVisibleDuringTearDown = e.getCause() == _cleanupFailure; + } + } + + public static void assertHealthyAfterSuite() { + assertCustomSuiteHealthy(); + } + } + + public static class LaterParticipant extends CustomSuiteHealthCheckedIntegrationTest { + @BeforeClass + public void prepareLaterClass() { + _laterClassSetupRan = true; + } + + @Test + public void laterClass() { + _laterClassRan = true; + } + } +} diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/suites/CustomClusterSuite.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CustomSuiteLifecycleCheckTest.java similarity index 51% copy from pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/suites/CustomClusterSuite.java copy to pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CustomSuiteLifecycleCheckTest.java index 1f4fc87548a..74a9e8a0aa4 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/suites/CustomClusterSuite.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CustomSuiteLifecycleCheckTest.java @@ -16,19 +16,34 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.pinot.integration.tests.suites; +package org.apache.pinot.integration.tests; -import org.junit.platform.suite.api.IncludeClassNamePatterns; -import org.junit.platform.suite.api.IncludeEngines; -import org.junit.platform.suite.api.SelectPackages; -import org.junit.platform.suite.api.Suite; +import org.testng.annotations.Test; -/// Runs the custom-data scenarios with one shared cluster. -/// Stateless suite definition; the selected tests run sequentially in one fork. -@Suite -@IncludeEngines("testng") -@SelectPackages("org.apache.pinot.integration.tests.custom") -@IncludeClassNamePatterns(".*") -public class CustomClusterSuite { +/// Validates cleanup after partial custom-suite startup and teardown failures without a cluster. +public class CustomSuiteLifecycleCheckTest { + @Test + public void testKafkaInitFailure() + throws Throwable { + CustomSuiteLifecycleProbe.verifyKafkaFailure("init"); + } + + @Test + public void testKafkaStartFailure() + throws Throwable { + CustomSuiteLifecycleProbe.verifyKafkaFailure("start"); + } + + @Test + public void testPartialStartupFailure() + throws Throwable { + CustomSuiteLifecycleProbe.verifyStartupFailure(); + } + + @Test + public void testTeardownFailure() + throws Throwable { + CustomSuiteLifecycleProbe.verifyTeardownFailure(); + } } diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CustomSuiteLifecycleProbe.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CustomSuiteLifecycleProbe.java new file mode 100644 index 00000000000..825ff297716 --- /dev/null +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CustomSuiteLifecycleProbe.java @@ -0,0 +1,269 @@ +/** + * 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.util.ArrayList; +import java.util.List; +import java.util.Properties; +import org.apache.pinot.broker.broker.helix.BaseBrokerStarter; +import org.apache.pinot.integration.tests.custom.CustomDataQueryClusterIntegrationTest; +import org.apache.pinot.plugin.stream.kafka30.server.EmbeddedKafkaCluster; +import org.apache.pinot.server.starter.helix.BaseServerStarter; +import org.apache.pinot.spi.data.Schema; + +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertSame; +import static org.testng.Assert.assertTrue; + + +/// Exercises custom-suite ownership and cleanup with in-memory starter doubles. +final class CustomSuiteLifecycleProbe { + private CustomSuiteLifecycleProbe() { + } + + static void verifyKafkaFailure(String phase) + throws Throwable { + FakeOwner owner = new FakeOwner(phase, false); + owner.resetCustomSuiteHealthBeforeSuite(); + Throwable failure = capture(owner::setUpSuite); + assertSame(failure, owner._startupFailure); + List<String> expected = phase.equals("init") + ? List.of("zk-start", "kafka-init", "kafka-stop", "zk-stop") + : List.of("zk-start", "kafka-init", "kafka-start", "kafka-stop", "zk-stop"); + assertEquals(owner._events, expected); + owner.tearDownSuite(); + owner.tearDownSuite(); + assertEquals(owner._events.get(owner._events.size() - 1), "zk-stop"); + assertEquals(owner._events.size(), expected.size(), "Teardown must not stop resources twice"); + owner.resetCustomSuiteHealthAfterSuite(); + } + + static void verifyStartupFailure() + throws Throwable { + FakeOwner owner = new FakeOwner("minion", true); + owner.resetCustomSuiteHealthBeforeSuite(); + Throwable failure = capture(owner::setUpSuite); + assertSame(failure, owner._startupFailure); + assertEquals(owner._events.subList(owner._events.size() - 7, owner._events.size()), + List.of("minion-stop", "server-1-stop", "server-2-stop", "broker-stop", "controller-stop", "kafka-stop", + "zk-stop")); + assertEquals(failure.getSuppressed().length, 2); + assertSame(failure.getSuppressed()[0], owner._stopFailure); + assertSame(failure.getSuppressed()[1], owner._brokerStopFailure); + assertEquals(owner._stopFailure.getSuppressed().length, 1); + assertSame(owner._stopFailure.getSuppressed()[0], owner._secondStopFailure); + int count = owner._events.size(); + owner.tearDownSuite(); + owner.tearDownSuite(); + assertEquals(owner._events.size(), count, "Teardown must not stop resources twice"); + owner.resetCustomSuiteHealthAfterSuite(); + } + + static void verifyTeardownFailure() + throws Throwable { + FakeOwner owner = new FakeOwner("none", true); + owner.resetCustomSuiteHealthBeforeSuite(); + owner.setUpSuite(); + Throwable failure = capture(owner::tearDownSuite); + assertSame(failure, owner._stopFailure); + assertEquals(failure.getSuppressed().length, 2); + assertSame(failure.getSuppressed()[0], owner._secondStopFailure); + assertSame(failure.getSuppressed()[1], owner._brokerStopFailure); + assertTrue(owner._events.contains("server-2-stop")); + assertTrue(owner._events.contains("broker-stop")); + assertTrue(owner._events.contains("controller-stop")); + assertTrue(owner._events.contains("kafka-stop")); + assertTrue(owner._events.contains("zk-stop")); + int count = owner._events.size(); + owner.tearDownSuite(); + assertEquals(owner._events.size(), count); + owner.resetCustomSuiteHealthAfterSuite(); + } + + private static Throwable capture(ThrowingAction action) { + try { + action.run(); + throw new AssertionError("Expected injected failure"); + } catch (Throwable failure) { + return failure; + } + } + + @FunctionalInterface + private interface ThrowingAction { + void run() + throws Throwable; + } + + private static final class FakeOwner extends CustomDataQueryClusterIntegrationTest { + private final List<String> _events = new ArrayList<>(); + private final RuntimeException _startupFailure = new RuntimeException("startup failure"); + private final RuntimeException _stopFailure = new RuntimeException("server stop failure"); + private final RuntimeException _secondStopFailure = new RuntimeException("second server stop failure"); + private final RuntimeException _brokerStopFailure = new RuntimeException("broker stop failure"); + private final String _phase; + private final boolean _failServerStop; + private final BaseBrokerStarter _broker = mock(BaseBrokerStarter.class); + private final BaseServerStarter _server1 = mock(BaseServerStarter.class); + private final BaseServerStarter _server2 = mock(BaseServerStarter.class); + private int _serverCount; + + private FakeOwner(String phase, boolean failServerStop) { + _phase = phase; + _failServerStop = failServerStop; + doAnswer(invocation -> { + _events.add("broker-stop"); + if (_failServerStop) { + throw _brokerStopFailure; + } + return null; + }).when(_broker).stop(); + doAnswer(invocation -> { + _events.add("server-1-stop"); + if (_failServerStop) { + throw _stopFailure; + } + return null; + }).when(_server1).stop(); + doAnswer(invocation -> { + _events.add("server-2-stop"); + if (_failServerStop) { + throw _secondStopFailure; + } + return null; + }).when(_server2).stop(); + } + + @Override + public void startZk() { + _events.add("zk-start"); + } + + @Override + public void stopZk() { + _events.add("zk-stop"); + } + + @Override + protected void startKafkaWithoutTopic() { + if (_phase.equals("init") || _phase.equals("start")) { + super.startKafkaWithoutTopic(); + } else { + _events.add("kafka-start"); + } + } + + @Override + protected EmbeddedKafkaCluster allocateCustomKafka() { + return new EmbeddedKafkaCluster() { + @Override + public void init(Properties props) { + _events.add("kafka-init"); + if (_phase.equals("init")) { + throw _startupFailure; + } + } + + @Override + public void start() { + _events.add("kafka-start"); + throw _startupFailure; + } + + @Override + public void stop() { + _events.add("kafka-stop"); + } + }; + } + + @Override + public void startController() { + _events.add("controller-start"); + } + + @Override + protected void startBroker() { + _events.add("broker-start"); + createBrokerStarter(); + } + + @Override + protected BaseBrokerStarter allocateCustomBroker() { + return _broker; + } + + @Override + protected void startServer() { + _events.add("server-start"); + createServerStarter(); + createServerStarter(); + } + + @Override + protected BaseServerStarter allocateCustomServer() { + return _serverCount++ == 0 ? _server1 : _server2; + } + + @Override + protected void startMinion() { + _events.add("minion-start"); + if (_phase.equals("minion")) { + throw _startupFailure; + } + } + + @Override + protected void cleanUpCustomMinion() { + _events.add("minion-stop"); + } + + @Override + protected void cleanUpCustomController() { + _events.add("controller-stop"); + } + + @Override + protected void cleanUpCustomKafka() { + if (_phase.equals("init") || _phase.equals("start")) { + super.cleanUpCustomKafka(); + } else { + _events.add("kafka-stop"); + } + } + + @Override + public String getTableName() { + return "fake-owner"; + } + + @Override + public Schema createSchema() { + return null; + } + + @Override + public List<File> createAvroFiles() { + return List.of(); + } + } +} diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java index b77c6fcef7e..df2282f1a7a 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java @@ -51,6 +51,7 @@ import org.testng.annotations.Test; /// Test for advanced push types. /// Currently only tests METADATA push type. /// todo: add test for URI push +@Test(suiteName = "CustomClusterIntegrationTest") public class SegmentUploadIntegrationTest extends SharedOfflineUploadIntegrationTestSuite { private static String _tableNameSuffix; diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentWriterUploaderIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentWriterUploaderIntegrationTest.java index 0f8b86c6d19..68b395797e2 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentWriterUploaderIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentWriterUploaderIntegrationTest.java @@ -48,6 +48,7 @@ import org.testng.annotations.Test; /// Tests creating segments via the [SegmentWriter] implementations +@Test(suiteName = "CustomClusterIntegrationTest") public class SegmentWriterUploaderIntegrationTest extends SharedOfflineUploadIntegrationTestSuite { private static final Logger LOGGER = LoggerFactory.getLogger(SegmentWriterUploaderIntegrationTest.class); diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SharedOfflineUploadIntegrationTestSuite.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SharedOfflineUploadIntegrationTestSuite.java index bfba726d38c..3c714d31c16 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SharedOfflineUploadIntegrationTestSuite.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SharedOfflineUploadIntegrationTestSuite.java @@ -18,13 +18,14 @@ */ 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.broker.broker.helix.BaseBrokerStarter; 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; @@ -36,62 +37,97 @@ import org.testng.annotations.AfterClass; import org.testng.annotations.AfterMethod; import org.testng.annotations.AfterSuite; import org.testng.annotations.BeforeClass; -import org.testng.annotations.BeforeMethod; -import org.testng.annotations.BeforeSuite; -/// Shares one default offline cluster between sequential segment-upload tests. -/// The owner has independent process state; each class retains its own data directories and configuration hooks. -/// Direct class selection uses the same suite callbacks to start and close its own cluster. -/// Parallel classes are rejected. -public abstract class SharedOfflineUploadIntegrationTestSuite extends BaseClusterIntegrationTest { +/// 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 OfflineUploadClusterOwner _owner; private static SharedOfflineUploadIntegrationTestSuite _activeClass; - private static Throwable _cleanupFailure; private final Set<String> _tables = new LinkedHashSet<>(); private final Set<String> _schemas = new LinkedHashSet<>(); - @BeforeSuite(alwaysRun = true) - public final void setUpSharedOfflineUploadSuite() - throws Throwable { - synchronized (SharedOfflineUploadIntegrationTestSuite.class) { - checkCleanupFailure(); - if (_owner == null) { - _owner = new OfflineUploadClusterOwner(); - try { - _owner.startCluster(); - } catch (Throwable t) { - _cleanupFailure = t; - Throwable cleanupFailure = _owner.stopCluster(null); - if (cleanupFailure != null) { - t.addSuppressed(cleanupFailure); - } - throw t; - } - } - } - } - @BeforeClass(alwaysRun = true) public final void bindSharedOfflineUploadCluster() { synchronized (SharedOfflineUploadIntegrationTestSuite.class) { - checkCleanupFailure(); + 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 = owner().getControllerConfig(); - _helixResourceManager = owner().getHelixResourceManager(); + _controllerConfig = getSharedControllerStarter().getConfig(); + _controllerBaseApiUrl = _controllerConfig.generateVipUrl(); + _helixResourceManager = getSharedHelixResourceManager(); + initControllerRequestURLBuilder(); } } - @BeforeMethod(alwaysRun = true) - public final void checkSharedOfflineUploadCleanup() { - checkCleanupFailure(); + // 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 @@ -142,7 +178,7 @@ public abstract class SharedOfflineUploadIntegrationTestSuite extends BaseCluste private Throwable cleanUpTables(Throwable failure) { for (String table : List.copyOf(_tables)) { try { - owner().removeTable(table); + removeTable(table); _tables.remove(table); } catch (Throwable t) { failure = appendFailure(failure, t); @@ -157,8 +193,8 @@ public abstract class SharedOfflineUploadIntegrationTestSuite extends BaseCluste if (_tables.isEmpty()) { for (String schema : List.copyOf(_schemas)) { try { - if (owner().getHelixResourceManager().getSchema(schema) != null) { - owner().deleteSchema(schema); + if (getSharedHelixResourceManager().getSchema(schema) != null) { + deleteSchema(schema); } _schemas.remove(schema); } catch (Throwable t) { @@ -178,7 +214,8 @@ public abstract class SharedOfflineUploadIntegrationTestSuite extends BaseCluste _pinotConnectionV2 = null; }); } - if (_tables.isEmpty() && _schemas.isEmpty()) { + 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; @@ -187,41 +224,20 @@ public abstract class SharedOfflineUploadIntegrationTestSuite extends BaseCluste @AfterSuite(alwaysRun = true) public final void tearDownSharedOfflineUploadSuite() throws Throwable { - synchronized (SharedOfflineUploadIntegrationTestSuite.class) { - if (_owner == null) { - return; - } - Throwable failure = null; - for (SharedOfflineUploadIntegrationTestSuite participant : PARTICIPANTS) { - failure = participant.cleanUpClass(failure); - } - failure = _owner.stopCluster(failure); - _owner = null; - _activeClass = null; - PARTICIPANTS.clear(); - if (failure != null) { - recordCleanupFailure(failure); - throw failure; - } - } - } - - private static OfflineUploadClusterOwner owner() { - if (_owner == null) { - throw new IllegalStateException("The shared offline upload cluster has not started"); + Throwable failure = null; + for (SharedOfflineUploadIntegrationTestSuite participant : PARTICIPANTS) { + failure = participant.cleanUpClass(failure); } - return _owner; - } - - private static void checkCleanupFailure() { - if (_cleanupFailure != null) { - throw new IllegalStateException("The shared offline upload cluster did not clean up successfully", - _cleanupFailure); + PARTICIPANTS.clear(); + _activeClass = null; + if (failure != null) { + recordCleanupFailure(failure); + throw failure; } } - private static synchronized void recordCleanupFailure(Throwable failure) { - _cleanupFailure = appendFailure(_cleanupFailure, failure); + private static void recordCleanupFailure(Throwable failure) { + recordCustomSuiteCleanupFailure(failure); } private static Throwable appendFailure(Throwable failure, Throwable next) { @@ -253,105 +269,39 @@ public abstract class SharedOfflineUploadIntegrationTestSuite extends BaseCluste @Override public final PinotAdminClient getOrCreateAdminClient() throws IOException { - return owner().getOrCreateAdminClient(); - } - - @Override - public final String getControllerBaseApiUrl() { - return owner().getControllerBaseApiUrl(); - } - - @Override - public final String getZkUrl() { - return owner().getZkUrl(); - } - - @Override - public final String getHelixClusterName() { - return owner().getHelixClusterName(); - } - - @Override - protected final String getBrokerBaseApiUrl() { - return owner().getBrokerBaseApiUrl(); - } - - @Override - protected final String getBrokerGrpcEndpoint() { - return owner().getBrokerGrpcEndpoint(); - } - - /// Owns only the default ZK/controller/broker/server processes; no test class owns these resources. - private static final class OfflineUploadClusterOwner extends BaseClusterIntegrationTest { - private BaseBrokerStarter _allocatedBroker; - private BaseServerStarter _allocatedServer; - - private void startCluster() - throws Exception { - startZk(); - startController(); - startBroker(); - startServer(); + if (_sharedClusterTestSuite == this) { + return super.getOrCreateAdminClient(); } + return _sharedClusterTestSuite.getOrCreateAdminClient(); + } - @Override - protected BaseBrokerStarter createBrokerStarter() { - _allocatedBroker = super.createBrokerStarter(); - return _allocatedBroker; + private void removeTable(String tableNameWithType) + throws IOException { + List<BaseServerStarter> servers = getSharedServerStarters(); + if (servers.size() != 2) { + throw new IllegalStateException("Expected two custom suite servers while cleaning up an offline table"); } - - @Override - protected BaseServerStarter createServerStarter() { - _allocatedServer = super.createServerStarter(); - return _allocatedServer; + if (getSharedHelixResourceManager().getOfflineTableConfig(tableNameWithType) != null) { + dropOfflineTable(tableNameWithType); } - - private Throwable stopCluster(Throwable failure) { - if (_allocatedServer != null) { - // startServers registers a starter only after start succeeds; retain partially initialized starters too. - if (!_serverStarters.contains(_allocatedServer)) { - _serverStarters.add(_allocatedServer); - } - failure = cleanUp(failure, () -> { - stopServer(); - _allocatedServer = null; - }); - } - if (_allocatedBroker != null) { - if (!_brokerStarters.contains(_allocatedBroker)) { - _brokerStarters.add(_allocatedBroker); + TestUtils.waitForCondition(ignored -> { + for (BaseServerStarter server : servers) { + if (server.getServerInstance().getInstanceDataManager().getTableDataManager(tableNameWithType) != null) { + return false; } - failure = cleanUp(failure, () -> { - stopBroker(); - _allocatedBroker = null; - }); - } - if (_controllerStarter != null) { - failure = cleanUp(failure, this::stopController); - } - return cleanUp(failure, this::stopZk); - } - - private void removeTable(String tableNameWithType) - throws IOException { - if (_serverStarters.size() != 1) { - throw new IllegalStateException("Expected one server while cleaning up an offline upload table"); } - if (_helixResourceManager.getOfflineTableConfig(tableNameWithType) != null) { - dropOfflineTable(tableNameWithType); + return true; + }, 60_000L, "Failed to remove table data manager for table: " + tableNameWithType); + waitForEVToDisappear(tableNameWithType); + String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType); + TestUtils.waitForCondition(ignored -> { + try { + getDebugInfo("debug/routingTable/" + rawTableName); + return false; + } catch (Exception e) { + String message = e.getMessage(); + return message != null && message.contains("Got error status code: 404"); } - waitForTableDataManagerRemoved(tableNameWithType); - waitForEVToDisappear(tableNameWithType); - String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType); - TestUtils.waitForCondition(ignored -> { - try { - getDebugInfo("debug/routingTable/" + rawTableName); - return false; - } catch (Exception e) { - String message = e.getMessage(); - return message != null && message.contains("Got error status code: 404"); - } - }, 60_000L, "Routing table still exists after dropping offline upload table " + rawTableName); - } + }, 60_000L, "Routing table still exists after dropping offline upload table " + rawTableName); } } diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomDataQueryClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomDataQueryClusterIntegrationTest.java index f6096b7a377..140283ac274 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomDataQueryClusterIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomDataQueryClusterIntegrationTest.java @@ -24,6 +24,7 @@ import java.io.IOException; import java.time.Duration; import java.util.ArrayList; import java.util.HashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Properties; @@ -39,13 +40,14 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.pinot.broker.broker.helix.BaseBrokerStarter; import org.apache.pinot.controller.BaseControllerStarter; import org.apache.pinot.controller.helix.core.PinotHelixResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotHelixTaskResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotTaskManager; -import org.apache.pinot.integration.tests.BaseClusterIntegrationTest; import org.apache.pinot.integration.tests.ClusterIntegrationTestUtils; import org.apache.pinot.plugin.stream.kafka.KafkaStreamConfigProperties; +import org.apache.pinot.plugin.stream.kafka30.server.EmbeddedKafkaCluster; import org.apache.pinot.server.starter.helix.BaseServerStarter; import org.apache.pinot.spi.config.table.TableConfig; import org.apache.pinot.spi.config.table.TableType; @@ -62,7 +64,7 @@ import org.testng.annotations.BeforeClass; import org.testng.annotations.BeforeSuite; -public abstract class CustomDataQueryClusterIntegrationTest extends BaseClusterIntegrationTest { +public abstract class CustomDataQueryClusterIntegrationTest extends CustomSuiteHealthCheckedIntegrationTest { protected static final Logger LOGGER = LoggerFactory.getLogger(CustomDataQueryClusterIntegrationTest.class); private static final int REALTIME_TABLE_CONFIG_RETRY_COUNT = 5; private static final long REALTIME_TABLE_CONFIG_RETRY_WAIT_MS = 1_000L; @@ -70,41 +72,198 @@ public abstract class CustomDataQueryClusterIntegrationTest extends BaseClusterI protected static CustomDataQueryClusterIntegrationTest _sharedClusterTestSuite = null; protected static final String TIMESTAMP_FIELD_NAME = "ts"; + private final List<BaseBrokerStarter> _allocatedBrokers = new ArrayList<>(); + private final List<BaseServerStarter> _allocatedServers = new ArrayList<>(); + private EmbeddedKafkaCluster _allocatedKafka; + private boolean _zkAttempted; + private boolean _kafkaAttempted; + private boolean _controllerAttempted; + private boolean _brokerAttempted; + private boolean _serverAttempted; + private boolean _minionAttempted; + private boolean _suiteCleanupAttempted; + + @Override + protected EmbeddedKafkaCluster createEmbeddedKafkaCluster() { + _allocatedKafka = allocateCustomKafka(); + return _allocatedKafka; + } + + @Override + protected BaseBrokerStarter createBrokerStarter() { + BaseBrokerStarter starter = allocateCustomBroker(); + _allocatedBrokers.add(starter); + return starter; + } + + @Override + protected BaseServerStarter createServerStarter() { + BaseServerStarter starter = allocateCustomServer(); + _allocatedServers.add(starter); + return starter; + } + + protected EmbeddedKafkaCluster allocateCustomKafka() { + return super.createEmbeddedKafkaCluster(); + } + + protected BaseBrokerStarter allocateCustomBroker() { + return super.createBrokerStarter(); + } + + protected BaseServerStarter allocateCustomServer() { + return super.createServerStarter(); + } + @BeforeSuite public void setUpSuite() - throws Exception { + throws Throwable { LOGGER.warn("Setting up integration test suite"); TestUtils.ensureDirectoriesExistAndEmpty(_tempDir, _segmentDir, _tarDir); _sharedClusterTestSuite = this; - - // Start the Pinot cluster - startZk(); - LOGGER.warn("Start Kafka in the integration test suite"); - startKafkaWithoutTopic(); - startController(); - startBroker(); - startServer(); - startMinion(); - LOGGER.warn("Finished setting up integration test suite"); + try { + _zkAttempted = true; + startZk(); + LOGGER.warn("Start Kafka in the integration test suite"); + _kafkaAttempted = true; + startKafkaWithoutTopic(); + _controllerAttempted = true; + startController(); + _brokerAttempted = true; + startBroker(); + _serverAttempted = true; + startServer(); + _minionAttempted = true; + startMinion(); + LOGGER.warn("Finished setting up integration test suite"); + } catch (Throwable failure) { + recordCustomSuiteCleanupFailure(failure); + cleanUpSuite(failure); + throw failure; + } } - @AfterSuite + @AfterSuite(alwaysRun = true) public void tearDownSuite() - throws Exception { + throws Throwable { + CustomDataQueryClusterIntegrationTest owner = _sharedClusterTestSuite; + if (owner == null) { + return; + } LOGGER.warn("Tearing down integration test suite"); - // Shutdown the Pinot cluster - stopMinion(); - stopServer(); - stopBroker(); - stopController(); - // Stop Kafka - LOGGER.warn("Stop Kafka in the integration test suite"); - stopKafka(); - stopZk(); - FileUtils.deleteDirectory(_tempDir); + Throwable failure = owner.cleanUpSuite(null); + _sharedClusterTestSuite = null; + if (failure != null) { + recordCustomSuiteCleanupFailure(failure); + throw failure; + } LOGGER.warn("Finished tearing down integration test suite"); } + private Throwable cleanUpSuite(Throwable failure) { + if (_suiteCleanupAttempted) { + return failure; + } + _suiteCleanupAttempted = true; + if (_minionAttempted) { + failure = cleanUp(failure, this::cleanUpCustomMinion); + } + if (_serverAttempted) { + failure = cleanUp(failure, this::cleanUpCustomServers); + } + if (_brokerAttempted) { + failure = cleanUp(failure, this::cleanUpCustomBrokers); + } + if (_controllerAttempted) { + failure = cleanUp(failure, this::cleanUpCustomController); + } + if (_kafkaAttempted) { + failure = cleanUp(failure, this::cleanUpCustomKafka); + } + if (_zkAttempted) { + failure = cleanUp(failure, this::stopZk); + } + return cleanUp(failure, () -> FileUtils.deleteDirectory(_tempDir)); + } + + protected void cleanUpCustomMinion() { + if (_minionStarter != null) { + stopMinion(); + } + } + + protected void cleanUpCustomServers() + throws Throwable { + Throwable failure = null; + LinkedHashSet<BaseServerStarter> starters = new LinkedHashSet<>(_serverStarters); + starters.addAll(_allocatedServers); + for (BaseServerStarter starter : starters) { + failure = cleanUp(failure, starter::stop); + } + if (failure != null) { + throw failure; + } + FileUtils.deleteQuietly(new File(TEMP_SERVER_DIR)); + _serverStarters.clear(); + _allocatedServers.clear(); + _serverGrpcPort = 0; + _serverAdminApiPort = 0; + _serverNettyPort = 0; + } + + protected void cleanUpCustomBrokers() + throws Throwable { + Throwable failure = null; + LinkedHashSet<BaseBrokerStarter> starters = new LinkedHashSet<>(_brokerStarters); + starters.addAll(_allocatedBrokers); + for (BaseBrokerStarter starter : starters) { + failure = cleanUp(failure, starter::stop); + } + if (failure != null) { + throw failure; + } + _brokerStarters.clear(); + _allocatedBrokers.clear(); + _brokerPorts.clear(); + _brokerBaseApiUrl = null; + _brokerGrpcEndpoint = null; + } + + protected void cleanUpCustomController() { + if (_controllerStarter != null) { + stopController(); + } + } + + protected void cleanUpCustomKafka() { + if (_kafkaStarters != null) { + stopKafka(); + } else if (_allocatedKafka != null) { + // Startup can fail before the shared starter list is populated. + _allocatedKafka.stop(); + } + _allocatedKafka = null; + } + + private static Throwable cleanUp(Throwable failure, CleanupAction action) { + try { + action.run(); + } catch (Throwable next) { + if (failure == null) { + failure = next; + } else if (failure != next) { + failure.addSuppressed(next); + } + } + return failure; + } + + @FunctionalInterface + private interface CleanupAction { + void run() + throws Throwable; + } + @BeforeClass public void setUp() throws Exception { @@ -405,11 +564,42 @@ public abstract class CustomDataQueryClusterIntegrationTest extends BaseClusterI return getTableName() + "-kafka"; } + /// Original cluster defaults for scenarios that prepare their own offline tables. + protected final Schema createDefaultSchema() + throws IOException { + return super.createSchema(); + } + + protected final TableConfig createDefaultOfflineTableConfig() { + return super.createOfflineTableConfig(); + } + + protected final String getDefaultSortedColumn() { + return super.getSortedColumn(); + } + + protected final List<String> getDefaultInvertedIndexColumns() { + return super.getInvertedIndexColumns(); + } + + protected final List<String> getDefaultNoDictionaryColumns() { + return super.getNoDictionaryColumns(); + } + + protected final List<String> getDefaultRangeIndexColumns() { + return super.getRangeIndexColumns(); + } + + protected final List<String> getDefaultBloomFilterColumns() { + return super.getBloomFilterColumns(); + } + @Override public abstract String getTableName(); @Override - public abstract Schema createSchema(); + public abstract Schema createSchema() + throws IOException; public abstract List<File> createAvroFiles() throws Exception; diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomSuiteHealthCheckedIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomSuiteHealthCheckedIntegrationTest.java new file mode 100644 index 00000000000..eb5d997c90f --- /dev/null +++ b/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() { + assertCustomSuiteHealthy(); + } + + @BeforeMethod(alwaysRun = true) + public final void checkCustomSuiteHealthBeforeMethod() { + assertCustomSuiteHealthy(); + } + + protected static synchronized void assertCustomSuiteHealthy() { + if (_customSuiteCleanupFailure != null) { + throw new IllegalStateException("A custom suite participant did not clean up successfully", + _customSuiteCleanupFailure); + } + } + + protected static synchronized void recordCustomSuiteCleanupFailure(Throwable failure) { + if (_customSuiteCleanupFailure == null) { + _customSuiteCleanupFailure = failure; + } else if (_customSuiteCleanupFailure != failure) { + _customSuiteCleanupFailure.addSuppressed(failure); + } + } + + private static synchronized void resetCustomSuiteHealth() { + _customSuiteCleanupFailure = null; + } +} diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/suites/CustomClusterSuite.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/suites/CustomClusterSuite.java index 1f4fc87548a..36148289902 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/suites/CustomClusterSuite.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/suites/CustomClusterSuite.java @@ -18,8 +18,11 @@ */ package org.apache.pinot.integration.tests.suites; +import org.apache.pinot.integration.tests.SegmentUploadIntegrationTest; +import org.apache.pinot.integration.tests.SegmentWriterUploaderIntegrationTest; import org.junit.platform.suite.api.IncludeClassNamePatterns; import org.junit.platform.suite.api.IncludeEngines; +import org.junit.platform.suite.api.SelectClasses; import org.junit.platform.suite.api.SelectPackages; import org.junit.platform.suite.api.Suite; @@ -29,6 +32,7 @@ import org.junit.platform.suite.api.Suite; @Suite @IncludeEngines("testng") @SelectPackages("org.apache.pinot.integration.tests.custom") +@SelectClasses({SegmentWriterUploaderIntegrationTest.class, SegmentUploadIntegrationTest.class}) @IncludeClassNamePatterns(".*") public class CustomClusterSuite { } diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/suites/SharedOfflineUploadSuite.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/suites/SharedOfflineUploadSuite.java deleted file mode 100644 index ebc6447a22a..00000000000 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/suites/SharedOfflineUploadSuite.java +++ /dev/null @@ -1,41 +0,0 @@ -/** - * 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.suites; - -import org.apache.pinot.common.utils.OrderedTestNGSuite; -import org.apache.pinot.integration.tests.SegmentUploadIntegrationTest; -import org.apache.pinot.integration.tests.SegmentWriterUploaderIntegrationTest; -import org.junit.platform.suite.api.ConfigurationParameter; -import org.junit.platform.suite.api.IncludeEngines; -import org.junit.platform.suite.api.SelectClasses; -import org.junit.platform.suite.api.Suite; - - -/// Runs the offline segment upload scenarios with one shared single-server cluster. -/// Stateless suite definition; the selected tests run sequentially in one fork. -@Suite -@IncludeEngines("testng") -@ConfigurationParameter(key = "testng.listeners", - value = "org.apache.pinot.integration.tests.suites.SharedOfflineUploadSuite") -@SelectClasses({ - SegmentWriterUploaderIntegrationTest.class, - SegmentUploadIntegrationTest.class -}) -public class SharedOfflineUploadSuite extends OrderedTestNGSuite { -} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
