stream2000 commented on code in PR #9558: URL: https://github.com/apache/hudi/pull/9558#discussion_r1311178312
########## hudi-utilities/src/test/java/org/apache/hudi/utilities/multitable/TestHoodieMultiTableServicesMain.java: ########## @@ -0,0 +1,304 @@ +/* + * + * * 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.hudi.utilities.multitable; + +import org.apache.hudi.client.SparkRDDReadClient; +import org.apache.hudi.client.SparkRDDWriteClient; +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.model.HoodieCleaningPolicy; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator; +import org.apache.hudi.common.testutils.HoodieCommonTestHarness; +import org.apache.hudi.common.testutils.HoodieTestDataGenerator; +import org.apache.hudi.common.testutils.HoodieTestUtils; +import org.apache.hudi.config.HoodieCleanConfig; +import org.apache.hudi.config.HoodieCompactionConfig; +import org.apache.hudi.config.HoodieIndexConfig; +import org.apache.hudi.config.HoodieLayoutConfig; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.index.HoodieIndex; +import org.apache.hudi.keygen.constant.KeyGeneratorOptions; +import org.apache.hudi.table.action.commit.SparkBucketIndexPartitioner; +import org.apache.hudi.table.storage.HoodieStorageLayout; +import org.apache.hudi.testutils.providers.SparkProvider; +import org.apache.hudi.utilities.HoodieCompactor; + +import org.apache.hadoop.fs.Path; +import org.apache.spark.HoodieSparkKryoRegistrar$; +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaRDD; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.SQLContext; +import org.apache.spark.sql.SparkSession; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.ExecutionException; + +import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors; + +class TestHoodieMultiTableServicesMain extends HoodieCommonTestHarness implements SparkProvider { + + private static final Logger LOG = LoggerFactory.getLogger(ArchiveTask.class); + + protected boolean initialized = false; + + private static SparkSession spark; + private static SQLContext sqlContext; + private static JavaSparkContext jsc; + private static HoodieSparkEngineContext context; + + protected transient HoodieTestDataGenerator dataGen = null; + + @BeforeEach + public void init() throws IOException, ExecutionException, InterruptedException { + boolean initialized = spark != null; + if (!initialized) { + SparkConf sparkConf = conf(); + HoodieSparkKryoRegistrar$.MODULE$.register(sparkConf); + SparkRDDReadClient.addHoodieSupport(sparkConf); + spark = SparkSession.builder().config(sparkConf).getOrCreate(); + sqlContext = spark.sqlContext(); + jsc = new JavaSparkContext(spark.sparkContext()); + context = new HoodieSparkEngineContext(jsc); + } + initPath(); + prepareData(); + } + + @Test + public void testRunAllServices() throws IOException, ExecutionException, InterruptedException { + HoodieMultiTableServicesMain.Config cfg = getHoodieMultiServiceConfig(); + cfg.batch = true; + HoodieTableMetaClient metaClient1 = getMetaClient("table1"); + HoodieTableMetaClient metaClient2 = getMetaClient("table2"); + HoodieMultiTableServicesMain main = new HoodieMultiTableServicesMain(jsc, cfg); + main.startServices(); + // Verify cleans + Assertions.assertEquals(1, metaClient1.reloadActiveTimeline().getCleanerTimeline().countInstants()); + Assertions.assertEquals(1, metaClient2.reloadActiveTimeline().getCleanerTimeline().countInstants()); + // Verify delta commits + Assertions.assertEquals(2, metaClient1.reloadActiveTimeline().getDeltaCommitTimeline().countInstants()); + Assertions.assertEquals(2, metaClient2.reloadActiveTimeline().getDeltaCommitTimeline().countInstants()); + // Verify replace commits + Assertions.assertEquals(1, metaClient1.reloadActiveTimeline().getCompletedReplaceTimeline().countInstants()); + Assertions.assertEquals(1, metaClient2.reloadActiveTimeline().getCompletedReplaceTimeline().countInstants()); + // Verify compactions, delta commits and replace commits + Assertions.assertEquals(4, metaClient1.reloadActiveTimeline().getCommitsTimeline().countInstants()); + Assertions.assertEquals(4, metaClient2.reloadActiveTimeline().getCommitsTimeline().countInstants()); + } + + @Test + public void testRunAllServicesForSingleTable() throws IOException, ExecutionException, InterruptedException { + HoodieMultiTableServicesMain.Config cfg = getHoodieMultiServiceConfig(); + HoodieTableMetaClient metaClient1 = getMetaClient("table1"); + cfg.batch = true; + cfg.basePath = Collections.singletonList(metaClient1.getBasePath()); + HoodieMultiTableServicesMain main = new HoodieMultiTableServicesMain(jsc, cfg); + main.startServices(); + // Verify cleans + Assertions.assertEquals(1, metaClient1.reloadActiveTimeline().getCleanerTimeline().countInstants()); + // Verify delta commits + Assertions.assertEquals(2, metaClient1.reloadActiveTimeline().getDeltaCommitTimeline().countInstants()); + // Verify replace commits + Assertions.assertEquals(1, metaClient1.reloadActiveTimeline().getCompletedReplaceTimeline().countInstants()); + // Verify compactions, delta commits and replace commits + Assertions.assertEquals(4, metaClient1.reloadActiveTimeline().getCommitsTimeline().countInstants()); + } + + @Test + public void testStreamRunAllServices() throws IOException, ExecutionException, InterruptedException { + HoodieMultiTableServicesMain.Config cfg = getHoodieMultiServiceConfig(); + HoodieMultiTableServicesMain main = new HoodieMultiTableServicesMain(jsc, cfg); + new Thread(() -> { + try { + Thread.sleep(10000); + LOG.info("Shutdown the table services"); + main.cancel(); + } catch (InterruptedException e) { + // + } + }).start(); + main.startServices(); + HoodieTableMetaClient metaClient1 = getMetaClient("table1"); + HoodieTableMetaClient metaClient2 = getMetaClient("table2"); + // Verify cleans + Assertions.assertEquals(1, metaClient1.reloadActiveTimeline().getCleanerTimeline().countInstants()); + Assertions.assertEquals(1, metaClient2.reloadActiveTimeline().getCleanerTimeline().countInstants()); + // Verify compactions, delta commits and replace commits + Assertions.assertEquals(4, metaClient1.reloadActiveTimeline().getCommitsTimeline().countInstants()); + Assertions.assertEquals(4, metaClient2.reloadActiveTimeline().getCommitsTimeline().countInstants()); + } + + private void prepareData() throws IOException { + initTestDataGenerator(); + HoodieTableMetaClient metaClient1 = getMetaClient("table1"); + HoodieTableMetaClient metaClient2 = getMetaClient("table2"); + String instant1 = HoodieInstantTimeGenerator.createNewInstantTime(0); + writeToTable(metaClient1.getBasePath(), instant1, false); + writeToTable(metaClient2.getBasePath(), instant1, false); + String instant2 = HoodieInstantTimeGenerator.createNewInstantTime(1); + writeToTable(metaClient1.getBasePath(), instant2, true); + writeToTable(metaClient2.getBasePath(), instant2, true); + Assertions.assertEquals(0, metaClient1.reloadActiveTimeline().getCleanerTimeline().countInstants()); + Assertions.assertEquals(0, metaClient2.reloadActiveTimeline().getCleanerTimeline().countInstants()); + } + + private void writeToTable(String basePath, String instant, boolean update) throws IOException { + String tableName = "test"; + HoodieWriteConfig.Builder writeConfigBuilder = getWriteConfigBuilder(basePath, tableName); + // enable files and bloom_filters on the regular write client + HoodieWriteConfig writeConfig = writeConfigBuilder.build(); + // do one upsert with synchronous metadata update + SparkRDDWriteClient writeClient = new SparkRDDWriteClient(context, writeConfig); + List<HoodieRecord> records; + writeClient.startCommitWithTime(instant); + if (update) { + records = dataGen.generateUpdates(instant, 100); + } else { + records = dataGen.generateInserts(instant, 100); + } + JavaRDD<WriteStatus> result = writeClient.upsert(jsc.parallelize(records, 8), instant); + List<WriteStatus> statuses = result.collect(); + assertNoWriteErrors(statuses); + } + + private HoodieWriteConfig.Builder getWriteConfigBuilder(String basePath, String tableName) { + Properties properties = new Properties(); + properties.setProperty(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "_row_key"); + return HoodieWriteConfig.newBuilder() + .withPath(basePath) + .withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA) + .withParallelism(4, 4) + .withBulkInsertParallelism(4) + .withFinalizeWriteParallelism(2) + .withProps(makeIndexConfig(HoodieIndex.IndexType.BUCKET)) Review Comment: Just want to make it easier to produce log for compaciton. -- 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]
