voonhous commented on code in PR #13147: URL: https://github.com/apache/hudi/pull/13147#discussion_r3592939993
########## hudi-timeline-service/src/test/java/org/apache/hudi/timeline/service/TestUiApi.java: ########## @@ -0,0 +1,713 @@ +/* + * 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.timeline.service; + +import org.apache.hudi.avro.model.HoodieActionInstant; +import org.apache.hudi.avro.model.HoodieCleanerPlan; +import org.apache.hudi.avro.model.HoodieIndexCommitMetadata; +import org.apache.hudi.avro.model.HoodieIndexPartitionInfo; +import org.apache.hudi.common.config.HoodieCommonConfig; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieReplaceCommitMetadata; +import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.timeline.versioning.clean.CleanPlanV2MigrationHandler; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.FileSystemViewStorageConfig; +import org.apache.hudi.common.table.view.FileSystemViewStorageType; +import org.apache.hudi.common.testutils.HoodieCommonTestHarness; +import org.apache.hudi.common.testutils.HoodieTestTable; +import org.apache.hudi.common.testutils.HoodieTestUtils; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.hadoop.conf.Configuration; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.net.URL; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.UUID; + +import static org.apache.hudi.common.table.view.RemoteHoodieTableFileSystemView.BASEPATH_PARAM; +import static org.apache.hudi.common.table.view.RemoteHoodieTableFileSystemView.LAST_INSTANT_URL; +import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightCompaction; +import static org.apache.hudi.common.testutils.FileCreateUtils.createRequestedCleanFile; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests for the Timeline UI API (routes under {@code /ui} and {@code /ui/api}), gated behind + * {@code TimelineService.Config.enableUi}. Exercises the JSON contract consumed by the browser + * UI, the path-traversal defense on the instant-details route, and the enable-ui flag gating. + */ +class TestUiApi extends HoodieCommonTestHarness { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private static final String UI_TIMELINE_URL = "/ui/api/timeline/instants/all"; + private static final String UI_INSTANT_URL = "/ui/api/timeline/instant"; + private static final String UI_CONFIG_URL = "/ui/api/table/config"; + private static final String UI_SCHEMA_URL = "/ui/api/table/schema/history"; + private static final String UI_PAGE_URL = "/ui"; + private static final String UI_STATIC_JS_URL = "/ui/static/js/timeline.js"; + + private static final String INSTANT_PARAM = "instant"; + private static final String INSTANT_ACTION_PARAM = "instantaction"; + private static final String INSTANT_STATE_PARAM = "instantstate"; + private static final String LIMIT_PARAM = "limit"; + + // A minimal but valid Avro schema, so TableSchemaResolver can parse it when computing currentSchema. + private static final String SCHEMA_A = + "{\"type\":\"record\",\"name\":\"R\",\"fields\":[{\"name\":\"id\",\"type\":\"long\"}]}"; + private static final String SCHEMA_B = + "{\"type\":\"record\",\"name\":\"R\",\"fields\":[{\"name\":\"id\",\"type\":\"long\"}," + + "{\"name\":\"name\",\"type\":[\"null\",\"string\"],\"default\":null}]}"; + private static final String SCHEMA_C = + "{\"type\":\"record\",\"name\":\"R\",\"fields\":[{\"name\":\"id\",\"type\":\"long\"}," + + "{\"name\":\"name\",\"type\":[\"null\",\"string\"],\"default\":null}," + + "{\"name\":\"amount\",\"type\":[\"null\",\"double\"],\"default\":null}]}"; + + private Configuration configuration; + private TimelineService server; + private int port; + + @BeforeEach + void setUp() throws Exception { + configuration = new Configuration(); + server = startServer(true); + port = server.getServerPort(); + awaitUiReady(port); + } + + // Guards against a brief startup window where the freshly bound server can 404 a route. + private void awaitUiReady(int targetPort) throws InterruptedException { + long deadline = System.currentTimeMillis() + 10_000; + while (System.currentTimeMillis() < deadline) { + try { + if (httpGet(targetPort, UI_PAGE_URL, Collections.emptyMap()).code == 200) { + return; + } + } catch (IOException ignored) { + // server not yet accepting connections + } + Thread.sleep(50); + } + throw new IllegalStateException("UI server did not become ready on port " + targetPort); + } + + @AfterEach + void tearDown() { + if (server != null) { + server.close(); + } + } + + private TimelineService startServer(boolean enableUi) throws IOException { + FileSystemViewStorageConfig sConf = + FileSystemViewStorageConfig.newBuilder().withStorageType(FileSystemViewStorageType.SPILLABLE_DISK).build(); + HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder().build(); + HoodieCommonConfig commonConfig = HoodieCommonConfig.newBuilder().build(); + HoodieLocalEngineContext ctx = new HoodieLocalEngineContext(new HadoopStorageConfiguration(configuration)); + TimelineService svc = TimelineServiceTestHarness.newBuilder().build( + configuration, + TimelineService.Config.builder().serverPort(0).enableUi(enableUi).build(), + FileSystemViewManager.createViewManager(ctx, metadataConfig, sConf, commonConfig)); + svc.startService(); + return svc; + } + + // --------------------------------------------------------------------------- + // Test-table helpers + // --------------------------------------------------------------------------- + + private HoodieTableMetaClient initTable(String name) throws IOException { + return HoodieTestUtils.init(tempDir.resolve(name).toAbsolutePath().toString()); + } + + private HoodieCommitMetadata commitMetadata(HoodieTableMetaClient mc, String basePath, String ts, + Map<String, String> extra) throws IOException { + return getCommitMetadata(mc, basePath, "par", ts, 1, extra).get(); + } + + private HoodieCommitMetadata commitMetadataWithSchema(HoodieTableMetaClient mc, String basePath, String ts, + String schema) throws IOException { + Map<String, String> extra = new HashMap<>(); + extra.put(HoodieCommitMetadata.SCHEMA_KEY, schema); + return commitMetadata(mc, basePath, ts, extra); + } + + // Writes an EMPTY instant file (any requested/inflight extension) straight into the timeline + // directory for actions with no HoodieTestTable helper. The instants/all route only lists instants, + // and the production inflight files of plan-carrying actions are empty by design. + private void createEmptyInstantFile(HoodieTableMetaClient mc, String ts, String extension) + throws IOException { + Path timelineDir = Paths.get(mc.getTimelinePath().toUri().getPath()); + Files.createDirectories(timelineDir); + Files.createFile(timelineDir.resolve(ts + extension)); + } + + // --------------------------------------------------------------------------- + // HTTP helpers + // --------------------------------------------------------------------------- + + private static final class Http { + final int code; + final String body; + final String contentType; + + Http(int code, String body, String contentType) { + this.code = code; + this.body = body; + this.contentType = contentType; + } + } + + private Http httpGet(int targetPort, String path, Map<String, String> params) throws IOException { + StringBuilder url = new StringBuilder("http://localhost:";).append(targetPort).append(path); + boolean first = true; + for (Map.Entry<String, String> e : params.entrySet()) { + url.append(first ? '?' : '&'); + url.append(URLEncoder.encode(e.getKey(), StandardCharsets.UTF_8)) + .append('=') + .append(URLEncoder.encode(e.getValue(), StandardCharsets.UTF_8)); + first = false; + } + HttpURLConnection conn = (HttpURLConnection) new URL(url.toString()).openConnection(); + conn.setRequestMethod("GET"); + // Avoid reusing a keep-alive connection to a previously closed server on a recycled port. + conn.setRequestProperty("Connection", "close"); + int code = conn.getResponseCode(); + String contentType = conn.getContentType(); + InputStream is = code >= 400 ? conn.getErrorStream() : conn.getInputStream(); + String body = is == null ? "" : new String(is.readAllBytes(), StandardCharsets.UTF_8); + conn.disconnect(); + return new Http(code, body, contentType); + } + + private Map<String, String> params(String... kv) { + Map<String, String> m = new LinkedHashMap<>(); + for (int i = 0; i < kv.length; i += 2) { + m.put(kv[i], kv[i + 1]); + } + return m; + } + + private JsonNode getJsonOk(String path, Map<String, String> params) throws IOException { + Http r = httpGet(port, path, params); + assertEquals(200, r.code, r.body); + return MAPPER.readTree(r.body); + } + + private JsonNode findInstant(JsonNode root, String requestTs) { + for (JsonNode n : root.get("instants")) { + if (requestTs.equals(n.get("requestTs").asText())) { + return n; + } + } + return null; + } + + private static boolean isJsonNull(JsonNode node) { + return node == null || node.isNull(); + } + + // --------------------------------------------------------------------------- + // 1. getUiTimeline mapping and ordering + // --------------------------------------------------------------------------- + + @Test + void testUiTimelineMappingAndOrder() throws Exception { + HoodieTableMetaClient mc = initTable("mapping"); + String base = mc.getBasePath().toString(); + String commitTs = "20240101000001"; + String compactionTs = "20240101000002"; + String cleanTs = "20240101000003"; + String logCompactionTs = "20240101000004"; + String clusteringTs = "20240101000005"; + + HoodieTestTable table = HoodieTestTable.of(mc); + table.addCommit(commitTs, Option.of(commitMetadata(mc, base, commitTs, Collections.emptyMap()))); + // A pending (requested) compaction: not yet completed. + table.addRequestedCompaction(compactionTs); + // A clean: a non-foldable action whose comparableAction equals its action. + table.addClean(cleanTs); + // Two more pending instants that pin the remaining comparable-action folds + // (logcompaction -> deltacommit, clustering -> replacecommit). No HoodieTestTable helper writes a + // requested logcompaction and the clustering helper needs Avro metadata, so write empty requested + // instant files directly; the instants/all route never reads their content. + createEmptyInstantFile(mc, logCompactionTs, HoodieTimeline.REQUESTED_LOG_COMPACTION_EXTENSION); + createEmptyInstantFile(mc, clusteringTs, HoodieTimeline.REQUESTED_CLUSTERING_COMMIT_EXTENSION); + + JsonNode root = getJsonOk(UI_TIMELINE_URL, params(BASEPATH_PARAM, base)); + JsonNode instants = root.get("instants"); + assertNotNull(instants, root.toString()); + + // Completed plain commit: comparableAction == action == commit, completionTs populated. + JsonNode commit = findInstant(root, commitTs); + assertNotNull(commit, root.toString()); + assertEquals("commit", commit.get("action").asText()); + assertEquals("commit", commit.get("comparableAction").asText()); + assertEquals("COMPLETED", commit.get("state").asText()); + assertFalse(isJsonNull(commit.get("completionTs")), "completed commit must carry a completionTs"); + + // Pending compaction: action=compaction, comparableAction folds to commit, completionTs null. + JsonNode compaction = findInstant(root, compactionTs); + assertNotNull(compaction, root.toString()); + assertEquals("compaction", compaction.get("action").asText()); + assertEquals("commit", compaction.get("comparableAction").asText()); + assertEquals("REQUESTED", compaction.get("state").asText()); + assertTrue(isJsonNull(compaction.get("completionTs")), "pending compaction must have null completionTs"); + + // Clean: non-foldable, comparableAction == action. + JsonNode clean = findInstant(root, cleanTs); + assertNotNull(clean, root.toString()); + assertEquals("clean", clean.get("action").asText()); + assertEquals("clean", clean.get("comparableAction").asText()); + + // Pending logcompaction: action=logcompaction, comparableAction folds to deltacommit, completionTs null. + JsonNode logCompaction = findInstant(root, logCompactionTs); + assertNotNull(logCompaction, root.toString()); + assertEquals("logcompaction", logCompaction.get("action").asText()); + assertEquals("deltacommit", logCompaction.get("comparableAction").asText()); + assertEquals("REQUESTED", logCompaction.get("state").asText()); + assertTrue(isJsonNull(logCompaction.get("completionTs")), "pending logcompaction must have null completionTs"); + + // Pending clustering: action=clustering, comparableAction folds to replacecommit, completionTs null. + JsonNode clustering = findInstant(root, clusteringTs); + assertNotNull(clustering, root.toString()); + assertEquals("clustering", clustering.get("action").asText()); + assertEquals("replacecommit", clustering.get("comparableAction").asText()); + assertEquals("REQUESTED", clustering.get("state").asText()); + assertTrue(isJsonNull(clustering.get("completionTs")), "pending clustering must have null completionTs"); + + // Instants returned in timeline (request-time ascending) order. + String previous = ""; + for (JsonNode n : instants) { + String ts = n.get("requestTs").asText(); + assertTrue(ts.compareTo(previous) >= 0, "instants not in ascending request-time order: " + root); + previous = ts; + } + } + + @Test + void testUiTimelineCompletedCompactionFoldsToCommit() throws Exception { + HoodieTableMetaClient mc = initTable("completed-compaction"); + String base = mc.getBasePath().toString(); + String ts = "20240101000010"; + HoodieCommitMetadata meta = commitMetadata(mc, base, ts, Collections.emptyMap()); + // Writes .compaction.requested, .compaction.inflight and a completed .commit for the same instant. + HoodieTestTable.of(mc).addCompaction(ts, meta); + + JsonNode root = getJsonOk(UI_TIMELINE_URL, params(BASEPATH_PARAM, base)); + + int matches = 0; + for (JsonNode n : root.get("instants")) { + if (ts.equals(n.get("requestTs").asText())) { + matches++; + // The active-timeline layout filter folds the (requested, inflight, completed) triple; the + // completed file is a .commit, so a completed compaction surfaces as action=commit. + assertEquals("commit", n.get("action").asText(), "completed compaction must surface as commit"); + assertEquals("COMPLETED", n.get("state").asText()); + assertFalse(isJsonNull(n.get("completionTs"))); + } + assertFalse("compaction".equals(n.get("action").asText()) && ts.equals(n.get("requestTs").asText()), + "completed compaction must never surface as action=compaction"); + } + assertEquals(1, matches, "completed compaction must surface exactly once: " + root); + } + + @Test + void testUiTimelineEmptyTableReturnsEmptyList() throws Exception { + HoodieTableMetaClient mc = initTable("empty-timeline"); + String base = mc.getBasePath().toString(); + + JsonNode root = getJsonOk(UI_TIMELINE_URL, params(BASEPATH_PARAM, base)); + JsonNode instants = root.get("instants"); + assertNotNull(instants, root.toString()); + assertTrue(instants.isArray(), root.toString()); + assertEquals(0, instants.size(), root.toString()); + } + + // --------------------------------------------------------------------------- + // 2. getInstantDetails + // --------------------------------------------------------------------------- + + @Test + void testGetInstantDetailsCommitRoundTrip() throws Exception { Review Comment: Added `testGetInstantDetailsCompletedSavepoint` -- writes an inflight + completed savepoint via `HoodieTestTable.addSavepointCommit` and asserts the avro `HoodieSavepointMetadata` fields (`savepointedBy`, `partitionMetadata.*.savepointDataFile`) come back through the Map conversion. -- 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]
