This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 3d4a0ce6f3 [#14127] fix(lineage): apply total queue capacity across
sinks (#14128)
3d4a0ce6f3 is described below
commit 3d4a0ce6f3950842444cedd88973b840b4862f71
Author: Qi Yu <[email protected]>
AuthorDate: Fri Oct 23 17:57:08 2026 +0800
[#14127] fix(lineage): apply total queue capacity across sinks (#14128)
### What changes were proposed in this pull request?
Use the documented `gravitino.lineage.queueCapacity` as the total
capacity, with `sinkQueueCapacity` retained as a compatibility alias.
The documented key takes precedence when both are set. Validate the
capacity as positive.
Convert the total into the top-level listener queue capacity so each
isolated lineage sink gets `max(1, total / sinkCount)` slots. Apply
generated settings after copying lineage properties so the original
total cannot overwrite the divided value. Document rounding and the
legacy key.
### Why are the changes needed?
The generated per-sink settings are ignored by `EventListenerManager`.
Queues therefore use either the 3000-slot listener default or the
undivided total, causing incorrect memory allocation and backpressure
thresholds.
Fix: #14127
### Does this PR introduce _any_ user-facing change?
Both lineage queue capacity keys now control the documented total across
sinks (default 10000). The documented key wins when both are present.
Each queue retains a minimum of one slot; division remainders are
unused. Non-positive values are rejected. The existing 90% high
watermark mechanism is unchanged.
### How was this patch tested?
- Added regression tests and verified failures before the production
fix.
- All 32 lineage tests passed with `./gradlew :lineage:test -PskipITs`.
- Eight real-queue scenarios block sinks with latches and check the high
watermark boundary for single/multiple sinks, both keys, precedence,
defaults, rounding, and the one-slot minimum.
- Configuration tests cover alias resolution and invalid capacities.
- Module Spotless checks and `git diff --check` passed.
- No external lineage backend or Docker integration tests were run.
Co-authored-by: Jerry Shao <[email protected]>
---
docs/lineage/gravitino-server-lineage.md | 6 ++
.../apache/gravitino/lineage/LineageConfig.java | 12 ++-
.../gravitino/lineage/sink/LineageSinkManager.java | 22 ++---
.../gravitino/lineage/TestLineageConfig.java | 38 ++++++-
.../lineage/sink/TestLineageSinkManager.java | 109 +++++++++++++++++----
5 files changed, 152 insertions(+), 35 deletions(-)
diff --git a/docs/lineage/gravitino-server-lineage.md
b/docs/lineage/gravitino-server-lineage.md
index 14a1afa6a3..eba0c3358d 100644
--- a/docs/lineage/gravitino-server-lineage.md
+++ b/docs/lineage/gravitino-server-lineage.md
@@ -20,6 +20,12 @@ Gravitino server provides a pluggable lineage framework to
receive, process, and
| `gravitino.lineage.${sinkName}.sinkClass` | The name of the lineage sink
class which should implement `org.apache.gravitino.lineage.sink.LineageSink`
interface.
| (none)
| No |
| `gravitino.lineage.queueCapacity` | The total capacity of
lineage event queues. When there are multiple lineage sinks, each sink utilizes
an isolated event queue. The capacity of each queue is calculated by dividing
the value of `gravitino.lineage.queueCapacity` by the number of sinks. | 10000
| No |
+The queue capacity must be positive. Each sink receives
+`max(1, floor(queueCapacity / numberOfSinks))` queue slots. Any division
remainder is
+unused; when the total is smaller than the number of sinks, each sink still
receives
+one slot. The legacy key `gravitino.lineage.sinkQueueCapacity` is also
accepted;
+`gravitino.lineage.queueCapacity` takes precedence when both are configured.
+
## Lineage HTTP Source
Http source provides an endpoint which follows [OpenLineage API
spec](https://openlineage.io/apidocs/openapi/) to receive OpenLineage run
event. The following use example:
diff --git
a/lineage/src/main/java/org/apache/gravitino/lineage/LineageConfig.java
b/lineage/src/main/java/org/apache/gravitino/lineage/LineageConfig.java
index e27c3d98da..9899d27793 100644
--- a/lineage/src/main/java/org/apache/gravitino/lineage/LineageConfig.java
+++ b/lineage/src/main/java/org/apache/gravitino/lineage/LineageConfig.java
@@ -22,6 +22,7 @@ package org.apache.gravitino.lineage;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -39,6 +40,9 @@ public class LineageConfig extends Config {
public static final String LINEAGE_CONFIG_PREFIX = "gravitino.lineage.";
public static final String LINEAGE_CONFIG_SINKS = "sinks";
+ /** The total capacity shared by the isolated lineage sink queues. */
+ public static final String LINEAGE_QUEUE_CAPACITY = "queueCapacity";
+
public static final String LINEAGE_SINK_QUEUE_CAPACITY = "sinkQueueCapacity";
public static final String LINEAGE_CONFIG_SOURCE = "source";
public static final String LINEAGE_SOURCE_CLASS_NAME = "sourceClass";
@@ -75,10 +79,12 @@ public class LineageConfig extends Config {
.createWithDefault(LINEAGE_LOG_SINK_NAME);
public static final ConfigEntry<Integer> SINK_QUEUE_CAPACITY =
- new ConfigBuilder(LINEAGE_SINK_QUEUE_CAPACITY)
+ new ConfigBuilder(LINEAGE_QUEUE_CAPACITY)
+ .alternatives(ImmutableList.of(LINEAGE_SINK_QUEUE_CAPACITY))
.doc("The capacity of the total lineage sink queue")
.version(ConfigConstants.VERSION_0_9_0)
.intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
.createWithDefault(LINEAGE_SINK_QUEUE_CAPACITY_DEFAULT);
public LineageConfig(Map<String, String> properties) {
@@ -115,9 +121,7 @@ public class LineageConfig extends Config {
m.put(LINEAGE_CONFIG_SINKS, sinkString);
}
- if (!m.containsKey(LINEAGE_SINK_QUEUE_CAPACITY)) {
- m.put(LINEAGE_SINK_QUEUE_CAPACITY,
String.valueOf(get(SINK_QUEUE_CAPACITY)));
- }
+ m.put(LINEAGE_QUEUE_CAPACITY, String.valueOf(get(SINK_QUEUE_CAPACITY)));
String logClassConfigKey =
LineageConfig.LINEAGE_LOG_SINK_NAME + "." +
LineageConfig.LINEAGE_SINK_CLASS_NAME;
diff --git
a/lineage/src/main/java/org/apache/gravitino/lineage/sink/LineageSinkManager.java
b/lineage/src/main/java/org/apache/gravitino/lineage/sink/LineageSinkManager.java
index ea30fb1029..ae28bbe61a 100644
---
a/lineage/src/main/java/org/apache/gravitino/lineage/sink/LineageSinkManager.java
+++
b/lineage/src/main/java/org/apache/gravitino/lineage/sink/LineageSinkManager.java
@@ -20,13 +20,11 @@
package org.apache.gravitino.lineage.sink;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
import io.openlineage.server.OpenLineage.RunEvent;
import java.io.Closeable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.lineage.LineageConfig;
import org.apache.gravitino.listener.EventBus;
import org.apache.gravitino.listener.EventListenerManager;
@@ -64,26 +62,27 @@ public class LineageSinkManager implements Closeable {
static Map<String, String> transformToEventListenerConfigs(
List<String> sinks, Map<String, String> lineageConfigs) {
Map<String, String> eventListenerConfigs = new HashMap<>();
- eventListenerConfigs.putAll(generateEventListenerConfigs(sinks,
lineageConfigs));
eventListenerConfigs.putAll(lineageConfigs);
+ // Generated listener settings must override the lineage total queue
capacity.
+ eventListenerConfigs.putAll(generateEventListenerConfigs(sinks,
lineageConfigs));
return eventListenerConfigs;
}
private static Map<String, String> generateEventListenerConfigs(
List<String> sinks, Map<String, String> lineageConfigs) {
- HashMap eventListenerConfigs = new HashMap();
+ Map<String, String> eventListenerConfigs = new HashMap<>();
if (sinks.isEmpty()) {
return eventListenerConfigs;
}
- String queueCapacity =
lineageConfigs.get(LineageConfig.LINEAGE_SINK_QUEUE_CAPACITY);
- Preconditions.checkArgument(
- StringUtils.isNotBlank(queueCapacity), "Lineage sink queue capacity is
not set");
-
- int totalCapacity = Integer.parseInt(queueCapacity);
- Preconditions.checkArgument(totalCapacity > 0, "Lineage sink queue
capacity must be positive");
+ int totalCapacity = new
LineageConfig(lineageConfigs).get(LineageConfig.SINK_QUEUE_CAPACITY);
int capacityPerSink = Math.max(1, totalCapacity / sinks.size());
+ // All lineage sinks use isolated queues of equal size. The listener
manager reads the
+ // top-level queue capacity rather than a per-listener setting.
+ eventListenerConfigs.put(
+ EventListenerManager.GRAVITINO_EVENT_LISTENER_QUEUE_CAPACITY,
+ String.valueOf(capacityPerSink));
eventListenerConfigs.put(
EventListenerManager.GRAVITINO_EVENT_LISTENER_NAMES, String.join(",",
sinks));
sinks.forEach(
@@ -91,9 +90,6 @@ public class LineageSinkManager implements Closeable {
eventListenerConfigs.put(
sinkName + "." +
EventListenerManager.GRAVITINO_EVENT_LISTENER_CLASS,
LineageSinkEventListener.class.getName());
- eventListenerConfigs.put(
- sinkName + "." +
EventListenerManager.GRAVITINO_EVENT_LISTENER_QUEUE_CAPACITY,
- String.valueOf(capacityPerSink));
});
return eventListenerConfigs;
}
diff --git
a/lineage/src/test/java/org/apache/gravitino/lineage/TestLineageConfig.java
b/lineage/src/test/java/org/apache/gravitino/lineage/TestLineageConfig.java
index 26f52733b2..fc5a4f8085 100644
--- a/lineage/src/test/java/org/apache/gravitino/lineage/TestLineageConfig.java
+++ b/lineage/src/test/java/org/apache/gravitino/lineage/TestLineageConfig.java
@@ -24,9 +24,45 @@ import java.util.Map;
import org.apache.gravitino.lineage.sink.LineageLogSink;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
public class TestLineageConfig {
+ @Test
+ void testDefaultQueueCapacity() {
+ Assertions.assertEquals(
+ "10000", new
LineageConfig(ImmutableMap.of()).getSinkConfigs().get("queueCapacity"));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"queueCapacity,120", "sinkQueueCapacity,120"})
+ void testQueueCapacityKeys(String key, String value) {
+ Assertions.assertEquals(
+ value,
+ new LineageConfig(ImmutableMap.of(key,
value)).getSinkConfigs().get("queueCapacity"));
+ }
+
+ @Test
+ void testDocumentedQueueCapacityTakesPrecedence() {
+ Assertions.assertEquals(
+ "120",
+ new LineageConfig(ImmutableMap.of("queueCapacity", "120",
"sinkQueueCapacity", "240"))
+ .getSinkConfigs()
+ .get("queueCapacity"));
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "queueCapacity,0", "queueCapacity,-1", "queueCapacity,invalid",
+ "sinkQueueCapacity,0", "sinkQueueCapacity,-1", "sinkQueueCapacity,invalid"
+ })
+ void testInvalidQueueCapacity(String key, String value) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> new LineageConfig(ImmutableMap.of(key, value)).getSinkConfigs());
+ }
+
@Test
void testLineageSource() {
// default config with HTTP source
@@ -64,7 +100,7 @@ public class TestLineageConfig {
sinkConfigs.get(
LineageConfig.LINEAGE_LOG_SINK_NAME + "." +
LineageConfig.LINEAGE_SINK_CLASS_NAME);
Assertions.assertEquals(LineageLogSink.class.getName(), className);
- String capacity =
sinkConfigs.get(LineageConfig.LINEAGE_SINK_QUEUE_CAPACITY);
+ String capacity = sinkConfigs.get(LineageConfig.LINEAGE_QUEUE_CAPACITY);
Assertions.assertEquals(
LineageConfig.LINEAGE_SINK_QUEUE_CAPACITY_DEFAULT,
Integer.parseInt(capacity));
diff --git
a/lineage/src/test/java/org/apache/gravitino/lineage/sink/TestLineageSinkManager.java
b/lineage/src/test/java/org/apache/gravitino/lineage/sink/TestLineageSinkManager.java
index 26918eaf26..719fddfc7c 100644
---
a/lineage/src/test/java/org/apache/gravitino/lineage/sink/TestLineageSinkManager.java
+++
b/lineage/src/test/java/org/apache/gravitino/lineage/sink/TestLineageSinkManager.java
@@ -20,10 +20,13 @@
package org.apache.gravitino.lineage.sink;
import io.openlineage.server.OpenLineage.RunEvent;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import lombok.SneakyThrows;
import org.apache.gravitino.lineage.LineageConfig;
import org.apache.gravitino.listener.AsyncQueueListener;
@@ -33,10 +36,95 @@ import
org.apache.gravitino.listener.EventListenerPluginWrapper;
import org.apache.gravitino.listener.api.EventListenerPlugin;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.mockito.Mockito;
public class TestLineageSinkManager {
+ @ParameterizedTest
+ @CsvSource({
+ "queueCapacity,1,10,10",
+ "queueCapacity,2,20,10",
+ "queueCapacity,2,21,10",
+ "sinkQueueCapacity,1,10,10",
+ "sinkQueueCapacity,2,20,10",
+ "queueCapacity,3,2,1",
+ "default,2,10000,5000",
+ "both,2,20,10"
+ })
+ void testActualQueueHighWatermark(
+ String capacityKey, int sinkCount, int totalCapacity, int
capacityPerSink) throws Exception {
+ Map<String, String> configs = new HashMap<>();
+ List<String> sinkNames = new ArrayList<>();
+ for (int i = 0; i < sinkCount; i++) {
+ String name = "sink" + i;
+ sinkNames.add(name);
+ configs.put(name + ".sinkClass", BlockingLineageSink.class.getName());
+ }
+ configs.put("sinks", String.join(",", sinkNames));
+ if ("both".equals(capacityKey)) {
+ configs.put("queueCapacity", String.valueOf(totalCapacity));
+ configs.put("sinkQueueCapacity", "200");
+ } else if (!"default".equals(capacityKey)) {
+ configs.put(capacityKey, String.valueOf(totalCapacity));
+ }
+ LineageConfig config = new LineageConfig(configs);
+ List<BlockingLineageSink> blockedSinks = new ArrayList<>();
+ try (LineageSinkManager manager = new LineageSinkManager()) {
+ manager.initialize(config.sinks(), config.getSinkConfigs());
+ List<AsyncQueueListener> queues = new ArrayList<>();
+ for (EventListenerPlugin listener :
manager.eventBus().getEventListeners()) {
+ AsyncQueueListener queue = (AsyncQueueListener) listener;
+ queues.add(queue);
+ EventListenerPluginWrapper wrapper =
+ (EventListenerPluginWrapper) queue.getEventListeners().get(0);
+ blockedSinks.add(
+ (BlockingLineageSink)
+ ((LineageSinkEventListener)
wrapper.getUserEventListener()).lineageSink());
+ }
+ Assertions.assertEquals(sinkCount, queues.size());
+ manager.sink(getEvent());
+ for (BlockingLineageSink sink : blockedSinks) {
+ Assertions.assertTrue(sink.entered.await(10, TimeUnit.SECONDS));
+ }
+
+ int threshold = (int) (capacityPerSink * 0.9f);
+ for (int i = 0; i < threshold; i++) {
+ manager.sink(getEvent());
+ }
+ queues.forEach(queue -> Assertions.assertFalse(queue.isHighWatermark()));
+ Assertions.assertFalse(manager.isHighWatermark());
+
+ manager.sink(getEvent());
+ queues.forEach(queue -> Assertions.assertTrue(queue.isHighWatermark()));
+ Assertions.assertTrue(manager.isHighWatermark());
+ } finally {
+ blockedSinks.forEach(sink -> sink.release.countDown());
+ }
+ }
+
+ /** A sink that holds the first event so queue watermarks can be checked
deterministically. */
+ public static class BlockingLineageSink implements LineageSink {
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ /**
+ * Blocks until the test releases the sink or the dispatcher is stopped.
+ *
+ * @param event the lineage event
+ */
+ @Override
+ public void sink(RunEvent event) {
+ entered.countDown();
+ try {
+ release.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
@SneakyThrows
@Test
void testLineageSinkManager() {
@@ -72,19 +160,14 @@ public class TestLineageSinkManager {
Assertions.assertEquals(
"sink1,sink2",
configs.get(EventListenerManager.GRAVITINO_EVENT_LISTENER_NAMES));
+ Assertions.assertEquals("500", configs.get("queueCapacity"));
- Assertions.assertEquals(
- "500",
- configs.get("sink1." +
EventListenerManager.GRAVITINO_EVENT_LISTENER_QUEUE_CAPACITY));
Assertions.assertEquals(
LineageSinkEventListener.class.getName(),
configs.get("sink1." +
EventListenerManager.GRAVITINO_EVENT_LISTENER_CLASS));
Assertions.assertEquals("sink1", configs.get("sink1.name"));
Assertions.assertEquals("a", configs.get("sink1.a"));
- Assertions.assertEquals(
- "500",
- configs.get("sink2." +
EventListenerManager.GRAVITINO_EVENT_LISTENER_QUEUE_CAPACITY));
Assertions.assertEquals(
LineageSinkEventListener.class.getName(),
configs.get("sink2." +
EventListenerManager.GRAVITINO_EVENT_LISTENER_CLASS));
@@ -101,17 +184,9 @@ public class TestLineageSinkManager {
Map<String, String> eventListenerConfigs =
LineageSinkManager.transformToEventListenerConfigs(sinks,
lineageConfigs);
- sinks.forEach(
- sink -> {
- String capacityKey =
- sink + "." +
EventListenerManager.GRAVITINO_EVENT_LISTENER_QUEUE_CAPACITY;
- String actualCapacity = eventListenerConfigs.get(capacityKey);
-
- Assertions.assertEquals(
- "1",
- actualCapacity,
- String.format("Queue capacity for %s should be at least 1",
sink));
- });
+ Assertions.assertEquals(
+ "1",
+
eventListenerConfigs.get(EventListenerManager.GRAVITINO_EVENT_LISTENER_QUEUE_CAPACITY));
}
@Test
