voonhous commented on code in PR #19575:
URL: https://github.com/apache/hudi/pull/19575#discussion_r3863088852
##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestDistributedRegistry.java:
##########
@@ -184,6 +187,131 @@ public void testAddMetricsParallel() {
Assertions.assertEquals(finalExpectedSum, metricCounts.get(METRIC_1));
}
+ @Test
+ public void testSetThrowsOnExecutor() {
+ // Given: a registry registered to the spark context
+ String registryName = REGISTRY_NAME + "_testSetOnExecutor";
+ Registry registry = engineContext.getMetricRegistry("", registryName);
+
+ List<Integer> data = new ArrayList<>();
+ data.add(1);
+
+ // When/Then: set() invoked on an executor must fail - it is
non-commutative under accumulator merges.
+ // The UnsupportedOperationException thrown on the executor surfaces
wrapped in a SparkException.
+ assertFailsOnExecutorWith("DistributedRegistry.set() must not be called
from a Spark executor", () ->
+ engineContext.map(data, value -> {
+ registry.set(METRIC_1, value);
+ return null;
+ }, 1));
+ }
+
+ @Test
+ public void testReleaseThrowsOnExecutor() {
+ // Given: a registry registered to the spark context
+ String registryName = REGISTRY_NAME + "_testReleaseOnExecutor";
+ Registry registry = engineContext.getMetricRegistry("", registryName);
+
+ List<Integer> data = new ArrayList<>();
+ data.add(1);
+
+ // When/Then: release() invoked on an executor must fail - clamping and
eviction are order-dependent
+ // under accumulator merges. The UnsupportedOperationException surfaces
wrapped in a SparkException.
+ assertFailsOnExecutorWith("DistributedRegistry.release() must not be
called from a Spark executor", () ->
+ engineContext.map(data, value -> {
+ registry.release(Collections.singletonMap(METRIC_1, (long) value));
+ return null;
+ }, 1));
+ }
+
+ /**
+ * Asserts the job failed because the executor-side guard fired, not for
some unrelated reason such as a serialization error.
+ */
+ private static void assertFailsOnExecutorWith(String expectedMessage,
Executable executable) {
+ SparkException thrown = Assertions.assertThrows(SparkException.class,
executable);
+ StringBuilder chain = new StringBuilder();
+ for (Throwable t = thrown; t != null; t = t.getCause()) {
+ chain.append(t).append('\n');
+ if (t.getCause() == t) {
+ break;
+ }
+ }
+ Assertions.assertTrue(chain.toString().contains(expectedMessage),
+ "expected the executor-side guard to fail the job, got: " + chain);
+ }
+
+ @Test
+ public void testSetOnDriverSucceeds() {
+ // set() on the driver (no TaskContext) remains supported.
+ DistributedRegistry registry = new DistributedRegistry(REGISTRY_NAME +
"_testSetOnDriver");
+ registry.set(METRIC_1, 42);
+ Assertions.assertEquals(42, registry.getAllCounts().get(METRIC_1));
+ }
+
+ @Test
+ public void testGetMetricRegistryReplacesNonDistributedRegistry() {
Review Comment:
Could the test at least be renamed to what it proves (a stale
`LocalRegistry` under the key is replaced by a fresh `DistributedRegistry`), so
it does not claim coverage of the `!(instanceof)` fallback? The branch itself
can wait for the lifetime change.
--
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]
