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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
