You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "zhang-arvin (via GitHub)" <gi...@apache.org> on 2026/08/18 12:53:44 UTC
[PR] test: add robust testing to streaming ingestion integration tests (druid)
zhang-arvin opened a new pull request, #20060:
URL: https://github.com/apache/druid/pull/20060
### Description
Adds more robust testing to streaming ingestion integration tests as described in #10971.
### Changes
1. **New `FaultyStreamEventStreamGenerator`** — A `SyntheticStreamGenerator` subclass that can inject faulty data at a configurable ratio:
- `INVALID_JSON` — Malformed JSON bytes that cannot be parsed (e.g., `{"broken": }`)
- `NULL_FIELDS` — Valid JSON objects where all field values are `null`
- `EMPTY_JSON` — Empty JSON objects (`{}`)
- `MULTI_ROW` — JSON arrays containing multiple objects
- `EMPTY_STRING` — Completely empty strings
- `ALL_VALID` — No faults injected (baseline)
2. **New `KafkaStreamIngestionRobustnessTest`** — Six integration tests that verify the streaming ingestion pipeline handles faulty data gracefully:
- `test_supervisorHandlesInvalidJsonGracefully` — Supervisor stays healthy with 20% invalid JSON
- `test_supervisorHandlesNullFieldsGracefully` — Supervisor stays healthy with 20% null-field records
- `test_supervisorHandlesEmptyJsonGracefully` — Supervisor stays healthy with 20% empty JSON objects
- `test_supervisorHandlesMultiRowDataGracefully` — Supervisor stays healthy with 20% multi-row JSON arrays
- `test_supervisorHandlesEmptyStringGracefully` — Supervisor stays healthy with 20% empty strings
- `test_supervisorHandlesAllValidDataCorrectly` — Baseline test with all valid data
Each test verifies that:
- The supervisor remains healthy (`RUNNING` state)
- Valid records are ingested successfully
- The row count matches the expected valid records
Closes #10971.
<hr>
This PR has:
- [x] been self-reviewed.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] test: add robust testing to streaming ingestion integration tests (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20060:
URL: https://github.com/apache/druid/pull/20060#issuecomment-6257181279
This pull request has been marked as stale due to 60 days of inactivity.
It will be closed in 4 weeks if no further activity occurs. If you think
that's incorrect or this pull request should instead be reviewed, please simply
write any comment. Even if closed, you can still revive the PR at any time or
discuss it on the dev@druid.apache.org list.
Thank you for your contributions.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] test: add robust testing to streaming ingestion integration tests (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #20060:
URL: https://github.com/apache/druid/pull/20060#discussion_r3813157818
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaStreamIngestionRobustnessTest.java:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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.druid.testing.embedded.indexing;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.indexing.kafka.simulate.KafkaResource;
+import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
+import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.apache.druid.testing.embedded.tools.EventSerializer;
+import org.apache.druid.testing.embedded.tools.FaultyStreamEventStreamGenerator;
+import org.apache.druid.testing.embedded.tools.FaultyStreamEventStreamGenerator.DataVariant;
+import org.apache.druid.testing.embedded.tools.JsonEventSerializer;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Integration tests that verify the robustness of streaming ingestion when the stream contains
+ * faulty data such as invalid JSON, null/empty fields, multi-row data, and empty strings.
+ *
+ * <p>These tests ensure that the supervisor and indexing tasks remain healthy and continue
+ * to ingest valid data even when some records are malformed.
+ */
+public class KafkaStreamIngestionRobustnessTest extends StreamIndexTestBase
+{
+ private final KafkaResource kafkaServer = new KafkaResource();
+
+ private static final int EVENTS_PER_SECOND = 6;
+ private static final long CYCLE_PADDING_MS = 100;
+ private static final int TOTAL_SECONDS = 10;
+ private static final double FAULTY_RATIO = 0.2; // 20% faulty records
+
+ @Override
+ protected StreamIngestResource<?> getStreamIngestResource()
+ {
+ return kafkaServer;
+ }
+
+ /**
+ * Creates a standard Kafka supervisor for JSON input format.
+ */
+ private KafkaSupervisorSpec createJsonSupervisor(final String topic)
+ {
+ return createKafkaSupervisor(kafkaServer)
+ .build(dataSource, topic);
+ }
+
+ /**
+ * Publishes records to the topic using a generator that mixes valid and faulty data.
+ *
+ * @return the number of valid records published
+ */
+ private int publishMixedRecords(
+ final String topic,
+ final DataVariant variant,
+ final int totalSeconds,
+ final double faultyRatio
+ )
+ {
+ final EventSerializer serializer = new JsonEventSerializer(overlord.bindings().jsonMapper());
+ final FaultyStreamEventStreamGenerator generator = new FaultyStreamEventStreamGenerator(
+ serializer,
+ EVENTS_PER_SECOND,
+ CYCLE_PADDING_MS,
+ variant,
+ faultyRatio
+ );
+
+ final List<byte[]> validEvents = generator.generateEvents(totalSeconds);
+ final List<byte[]> allRecords = new ArrayList<>();
+ int validCount = 0;
+
+ for (int i = 0; i < validEvents.size(); i++) {
+ if (generator.isFaultyEvent(i)) {
+ allRecords.add(generator.generateFaultyBytes(i));
+ } else {
+ allRecords.add(validEvents.get(i));
+ validCount++;
+ }
+ }
+
+ kafkaServer.publishRecordsToTopic(topic, allRecords);
+ return validCount;
+ }
+
+ @Test
+ @Timeout(60)
+ public void test_supervisorHandlesInvalidJsonGracefully()
+ {
+ final String topic = IdUtils.getRandomId();
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ // Publish mixed records: 80% valid + 20% invalid JSON
+ final int validCount = publishMixedRecords(topic, DataVariant.INVALID_JSON, TOTAL_SECONDS, FAULTY_RATIO);
+
+ // Create and start the supervisor
+ final KafkaSupervisorSpec supervisor = createJsonSupervisor(topic);
+ cluster.callApi().postSupervisor(supervisor);
+
+ // Verify supervisor is healthy
+ verifySupervisorIsRunningHealthy(supervisor.getId());
Review Comment:
[P2] Health is checked before malformed records are consumed
The supervisor is checked before malformed records are consumed; later assertions only wait for valid rows. A supervisor can become unhealthy afterward and the test still passes. Check health after input processing completes.
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/tools/FaultyStreamEventStreamGenerator.java:
##########
@@ -0,0 +1,195 @@
+/*
+ * 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.druid.testing.embedded.tools;
+
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.joda.time.DateTime;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * A {@link SyntheticStreamGenerator} that can inject faulty data (invalid format, null/empty fields,
+ * multi-row data, etc.) into the stream at a configurable ratio. This is used to test the robustness
+ * of streaming ingestion pipelines.
+ *
+ * <p>The generator wraps a delegate {@link EventSerializer} and produces three categories of events:
+ * <ul>
+ * <li><b>Valid events</b> — normal Wikipedia-style events serialized by the delegate</li>
+ * <li><b>Faulty events</b> — events that may be invalid JSON, contain null/empty fields, or
+ * use multi-row format, depending on the configured {@link DataVariant}</li>
+ * </ul>
+ *
+ * <p>The ratio of faulty events is controlled by {@link #faultyRatio} (0.0–1.0), where 0.0 means
+ * all events are valid and 1.0 means all events are faulty.
+ */
+public class FaultyStreamEventStreamGenerator extends SyntheticStreamGenerator
+{
+ private static final Logger LOG = new Logger(FaultyStreamEventStreamGenerator.class);
+
+ /**
+ * The type of faulty data to inject into the stream.
+ */
+ public enum DataVariant
+ {
+ /** All events are valid (no faults injected). */
+ ALL_VALID,
+ /** Inject malformed JSON bytes that cannot be parsed. */
+ INVALID_JSON,
+ /** Inject valid JSON with null field values. */
+ NULL_FIELDS,
+ /** Inject empty JSON objects ({}). */
+ EMPTY_JSON,
+ /** Inject multi-row JSON arrays containing multiple objects. */
+ MULTI_ROW,
+ /** Inject completely empty strings. */
+ EMPTY_STRING
+ }
+
+ private final EventSerializer delegate;
+ private final DataVariant variant;
+ private final double faultyRatio;
+
+ /**
+ * Creates a new faulty stream generator.
+ *
+ * @param delegate the serializer used for valid events
+ * @param eventsPerSecond number of events per second
+ * @param cyclePaddingMs padding for cycle timing
+ * @param variant the type of faulty data to inject
+ * @param faultyRatio the ratio (0.0–1.0) of faulty events to inject
+ */
+ public FaultyStreamEventStreamGenerator(
+ final EventSerializer delegate,
+ final int eventsPerSecond,
+ final long cyclePaddingMs,
+ final DataVariant variant,
+ final double faultyRatio
+ )
+ {
+ super(delegate, eventsPerSecond, cyclePaddingMs);
+ this.delegate = delegate;
+ this.variant = variant;
+ this.faultyRatio = Math.max(0.0, Math.min(1.0, faultyRatio));
+ }
+
+ @Override
+ List<Pair<String, Object>> newEvent(final int row, final DateTime timestamp)
+ {
+ // newEvent() is called by the parent's generateEvents() method, which bypasses
+ // the delegate serializer and serializes the event itself. When the parent
+ // serializes the event, it will use the delegate serializer, which will produce
+ // valid JSON. Faulty data injection is handled by overriding the run() method
+ // to replace some events with faulty bytes.
+ final List<Pair<String, Object>> event = new ArrayList<>();
+ event.add(Pair.of("timestamp", "2021-01-01T00:00:00Z"));
+ event.add(Pair.of("page", "Test Page"));
+ event.add(Pair.of("language", "en"));
+ event.add(Pair.of("user", "test"));
+ event.add(Pair.of("unpatrolled", "true"));
+ event.add(Pair.of("newPage", "true"));
+ event.add(Pair.of("robot", "false"));
+ event.add(Pair.of("anonymous", "false"));
+ event.add(Pair.of("namespace", "article"));
+ event.add(Pair.of("continent", "North America"));
+ event.add(Pair.of("country", "United States"));
+ event.add(Pair.of("region", "Bay Area"));
+ event.add(Pair.of("city", "San Francisco"));
+ event.add(Pair.of("added", row));
+ event.add(Pair.of("deleted", 0));
+ event.add(Pair.of("delta", row));
+ return Collections.unmodifiableList(event);
+ }
+
+ /**
+ * Returns whether the event at the given index should be faulty based on the configured ratio.
+ */
+ public boolean isFaultyEvent(final int eventIndex)
+ {
+ if (variant == DataVariant.ALL_VALID) {
+ return false;
+ }
+ return ThreadLocalRandom.current().nextDouble() < faultyRatio;
Review Comment:
[P2] Fault injection can emit zero faulty records
Random sampling can emit zero faulty records, while the tests never assert fault coverage. Use deterministic selection or an injected RNG and assert the expected faulty count.
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/tools/FaultyStreamEventStreamGenerator.java:
##########
@@ -0,0 +1,195 @@
+/*
+ * 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.druid.testing.embedded.tools;
+
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.joda.time.DateTime;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * A {@link SyntheticStreamGenerator} that can inject faulty data (invalid format, null/empty fields,
+ * multi-row data, etc.) into the stream at a configurable ratio. This is used to test the robustness
+ * of streaming ingestion pipelines.
+ *
+ * <p>The generator wraps a delegate {@link EventSerializer} and produces three categories of events:
+ * <ul>
+ * <li><b>Valid events</b> — normal Wikipedia-style events serialized by the delegate</li>
+ * <li><b>Faulty events</b> — events that may be invalid JSON, contain null/empty fields, or
+ * use multi-row format, depending on the configured {@link DataVariant}</li>
+ * </ul>
+ *
+ * <p>The ratio of faulty events is controlled by {@link #faultyRatio} (0.0–1.0), where 0.0 means
+ * all events are valid and 1.0 means all events are faulty.
+ */
+public class FaultyStreamEventStreamGenerator extends SyntheticStreamGenerator
+{
+ private static final Logger LOG = new Logger(FaultyStreamEventStreamGenerator.class);
+
+ /**
+ * The type of faulty data to inject into the stream.
+ */
+ public enum DataVariant
+ {
+ /** All events are valid (no faults injected). */
+ ALL_VALID,
+ /** Inject malformed JSON bytes that cannot be parsed. */
+ INVALID_JSON,
+ /** Inject valid JSON with null field values. */
+ NULL_FIELDS,
+ /** Inject empty JSON objects ({}). */
+ EMPTY_JSON,
+ /** Inject multi-row JSON arrays containing multiple objects. */
+ MULTI_ROW,
+ /** Inject completely empty strings. */
+ EMPTY_STRING
+ }
+
+ private final EventSerializer delegate;
+ private final DataVariant variant;
+ private final double faultyRatio;
+
+ /**
+ * Creates a new faulty stream generator.
+ *
+ * @param delegate the serializer used for valid events
+ * @param eventsPerSecond number of events per second
+ * @param cyclePaddingMs padding for cycle timing
+ * @param variant the type of faulty data to inject
+ * @param faultyRatio the ratio (0.0–1.0) of faulty events to inject
+ */
+ public FaultyStreamEventStreamGenerator(
+ final EventSerializer delegate,
+ final int eventsPerSecond,
+ final long cyclePaddingMs,
+ final DataVariant variant,
+ final double faultyRatio
+ )
+ {
+ super(delegate, eventsPerSecond, cyclePaddingMs);
+ this.delegate = delegate;
+ this.variant = variant;
+ this.faultyRatio = Math.max(0.0, Math.min(1.0, faultyRatio));
+ }
+
+ @Override
+ List<Pair<String, Object>> newEvent(final int row, final DateTime timestamp)
+ {
+ // newEvent() is called by the parent's generateEvents() method, which bypasses
+ // the delegate serializer and serializes the event itself. When the parent
+ // serializes the event, it will use the delegate serializer, which will produce
+ // valid JSON. Faulty data injection is handled by overriding the run() method
+ // to replace some events with faulty bytes.
+ final List<Pair<String, Object>> event = new ArrayList<>();
+ event.add(Pair.of("timestamp", "2021-01-01T00:00:00Z"));
+ event.add(Pair.of("page", "Test Page"));
+ event.add(Pair.of("language", "en"));
+ event.add(Pair.of("user", "test"));
+ event.add(Pair.of("unpatrolled", "true"));
+ event.add(Pair.of("newPage", "true"));
+ event.add(Pair.of("robot", "false"));
+ event.add(Pair.of("anonymous", "false"));
+ event.add(Pair.of("namespace", "article"));
+ event.add(Pair.of("continent", "North America"));
+ event.add(Pair.of("country", "United States"));
+ event.add(Pair.of("region", "Bay Area"));
+ event.add(Pair.of("city", "San Francisco"));
+ event.add(Pair.of("added", row));
+ event.add(Pair.of("deleted", 0));
+ event.add(Pair.of("delta", row));
+ return Collections.unmodifiableList(event);
+ }
+
+ /**
+ * Returns whether the event at the given index should be faulty based on the configured ratio.
+ */
+ public boolean isFaultyEvent(final int eventIndex)
+ {
+ if (variant == DataVariant.ALL_VALID) {
+ return false;
+ }
+ return ThreadLocalRandom.current().nextDouble() < faultyRatio;
+ }
+
+ /**
+ * Generates a faulty byte array for the given variant.
+ */
+ public byte[] generateFaultyBytes(final int eventIndex)
+ {
+ switch (variant) {
+ case INVALID_JSON:
+ return "{\"broken\": }".getBytes(StandardCharsets.UTF_8);
+ case NULL_FIELDS:
+ return ("{"
+ + "\"timestamp\": null,"
+ + "\"page\": null,"
+ + "\"language\": null,"
+ + "\"user\": null,"
+ + "\"unpatrolled\": null,"
+ + "\"newPage\": null,"
+ + "\"robot\": null,"
+ + "\"anonymous\": null,"
+ + "\"namespace\": null,"
+ + "\"continent\": null,"
+ + "\"country\": null,"
+ + "\"region\": null,"
+ + "\"city\": null,"
+ + "\"added\": null,"
+ + "\"deleted\": null,"
+ + "\"delta\": null"
+ + "}")
+ .getBytes(StandardCharsets.UTF_8);
+ case EMPTY_JSON:
+ return "{}".getBytes(StandardCharsets.UTF_8);
+ case MULTI_ROW:
+ return ("["
+ + "{\"timestamp\":\"2021-01-01T00:00:00Z\",\"page\":\"Multi1\",\"language\":\"en\",\"user\":\"test\",\"unpatrolled\":\"true\",\"newPage\":\"true\",\"robot\":\"false\",\"anonymous\":\"false\",\"namespace\":\"article\",\"continent\":\"North America\",\"country\":\"United States\",\"region\":\"Bay Area\",\"city\":\"San Francisco\",\"added\":1,\"deleted\":0,\"delta\":1},"
+ + "{\"timestamp\":\"2021-01-01T00:00:01Z\",\"page\":\"Multi2\",\"language\":\"en\",\"user\":\"test\",\"unpatrolled\":\"true\",\"newPage\":\"true\",\"robot\":\"false\",\"anonymous\":\"false\",\"namespace\":\"article\",\"continent\":\"North America\",\"country\":\"United States\",\"region\":\"Bay Area\",\"city\":\"San Francisco\",\"added\":2,\"deleted\":0,\"delta\":2}"
+ + "\"]")
Review Comment:
[P2] MULTI_ROW payload is invalid and unsupported
The MULTI_ROW payload has an invalid JSON suffix, so it is rejected as malformed. Kafka's non-line-splittable JsonReader also treats a top-level array as one value, not separate rows. The test only waits for valid rows, so it can pass without testing multi-row ingestion.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org