You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by gi...@apache.org on 2022/06/24 20:53:58 UTC
[druid] branch master updated: Fix flaky KafkaIndexTaskTest. (#12657)
This is an automated email from the ASF dual-hosted git repository.
gian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new d5abd06b96 Fix flaky KafkaIndexTaskTest. (#12657)
d5abd06b96 is described below
commit d5abd06b9679d0223b8362855de5cb064ad76ef0
Author: Gian Merlino <gi...@gmail.com>
AuthorDate: Fri Jun 24 13:53:51 2022 -0700
Fix flaky KafkaIndexTaskTest. (#12657)
* Fix flaky KafkaIndexTaskTest.
The testRunTransactionModeRollback case had many race conditions. Most notably,
it would commit a transaction and then immediately check to see that the results
were *not* indexed. This is racey because it relied on the indexing thread being
slower than the test thread.
Now, the case waits for the transaction to be processed by the indexing thread
before checking the results.
* Changes from review.
---
.../druid/indexing/kafka/KafkaIndexTask.java | 2 -
.../druid/indexing/kafka/KafkaIndexTaskTest.java | 174 +++++++++++++--------
.../SeekableStreamIndexTaskTestBase.java | 8 +-
3 files changed, 110 insertions(+), 74 deletions(-)
diff --git a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java
index 623379aae5..e24072e5a9 100644
--- a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java
+++ b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java
@@ -38,7 +38,6 @@ public class KafkaIndexTask extends SeekableStreamIndexTask<Integer, Long, Kafka
{
private static final String TYPE = "index_kafka";
- private final KafkaIndexTaskIOConfig ioConfig;
private final ObjectMapper configMapper;
// This value can be tuned in some tests
@@ -65,7 +64,6 @@ public class KafkaIndexTask extends SeekableStreamIndexTask<Integer, Long, Kafka
getFormattedGroupId(dataSchema.getDataSource(), TYPE)
);
this.configMapper = configMapper;
- this.ioConfig = ioConfig;
Preconditions.checkArgument(
ioConfig.getStartSequenceNumbers().getExclusivePartitions().isEmpty(),
diff --git a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
index 2067db25ef..afe9c68efe 100644
--- a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
+++ b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
@@ -96,6 +96,7 @@ import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.guava.Sequences;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.java.util.common.parsers.CloseableIterator;
import org.apache.druid.java.util.emitter.EmittingLogger;
@@ -124,7 +125,6 @@ import org.apache.druid.query.scan.ScanQueryConfig;
import org.apache.druid.query.scan.ScanQueryEngine;
import org.apache.druid.query.scan.ScanQueryQueryToolChest;
import org.apache.druid.query.scan.ScanQueryRunnerFactory;
-import org.apache.druid.query.scan.ScanResultValue;
import org.apache.druid.query.spec.QuerySegmentSpec;
import org.apache.druid.query.timeseries.TimeseriesQuery;
import org.apache.druid.query.timeseries.TimeseriesQueryEngine;
@@ -141,7 +141,6 @@ import org.apache.druid.segment.join.NoopJoinableFactory;
import org.apache.druid.segment.loading.DataSegmentPusher;
import org.apache.druid.segment.loading.LocalDataSegmentPusher;
import org.apache.druid.segment.loading.LocalDataSegmentPusherConfig;
-import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
import org.apache.druid.segment.realtime.firehose.NoopChatHandlerProvider;
import org.apache.druid.segment.transform.ExpressionTransform;
import org.apache.druid.segment.transform.TransformSpec;
@@ -181,6 +180,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
@@ -252,11 +252,9 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
private Long maxTotalRows = null;
private Period intermediateHandoffPeriod = null;
- private AppenderatorsManager appenderatorsManager;
private String topic;
private List<ProducerRecord<byte[], byte[]>> records;
private final Set<Integer> checkpointRequestsHash = new HashSet<>();
- private RowIngestionMetersFactory rowIngestionMetersFactory;
private static List<ProducerRecord<byte[], byte[]>> generateRecords(String topic)
{
@@ -356,7 +354,6 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
topic = getTopicName();
records = generateRecords(topic);
reportsFile = File.createTempFile("KafkaIndexTaskTestReports-" + System.currentTimeMillis(), "json");
- appenderatorsManager = new TestAppenderatorsManager();
makeToolboxFactory();
}
@@ -1289,7 +1286,7 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
new StringDimensionSchema("kafka.topic"),
new LongDimensionSchema("kafka.offset"),
new StringDimensionSchema("kafka.header.encoding")
- )
+ )
),
new AggregatorFactory[]{
new DoubleSumAggregatorFactory("met1sum", "met1"),
@@ -1324,13 +1321,12 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
final QuerySegmentSpec interval = OBJECT_MAPPER.readValue(
"\"2008/2012\"", QuerySegmentSpec.class
);
- List<ScanResultValue> scanResultValues = scanData(task, interval);
+ List<Map<String, Object>> scanResultValues = scanData(task, interval);
//verify that there are no records indexed in the rollbacked time period
Assert.assertEquals(3, Iterables.size(scanResultValues));
int i = 0;
- for (ScanResultValue result : scanResultValues) {
- final Map<String, Object> event = ((List<Map<String, Object>>) result.getEvents()).get(0);
+ for (Map<String, Object> event : scanResultValues) {
Assert.assertEquals((long) i++, event.get("kafka.offset"));
Assert.assertEquals(topic, event.get("kafka.topic"));
Assert.assertEquals("application/json", event.get("kafka.header.encoding"));
@@ -1401,13 +1397,11 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
final QuerySegmentSpec interval = OBJECT_MAPPER.readValue(
"\"2008/2012\"", QuerySegmentSpec.class
);
- List<ScanResultValue> scanResultValues = scanData(task, interval);
- //verify that there are no records indexed in the rollbacked time period
+ List<Map<String, Object>> scanResultValues = scanData(task, interval);
Assert.assertEquals(3, Iterables.size(scanResultValues));
int i = 0;
- for (ScanResultValue result : scanResultValues) {
- final Map<String, Object> event = ((List<Map<String, Object>>) result.getEvents()).get(0);
+ for (Map<String, Object> event : scanResultValues) {
Assert.assertEquals("application/json", event.get("kafka.testheader.encoding"));
Assert.assertEquals("y", event.get("dim2"));
}
@@ -2572,7 +2566,9 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
0,
"sequence0",
new SeekableStreamStartSequenceNumbers<>(topic, ImmutableMap.of(0, 0L), ImmutableSet.of()),
- new SeekableStreamEndSequenceNumbers<>(topic, ImmutableMap.of(0, 13L)),
+
+ // End offset is one after 12 real messages + 2 txn control messages (last seen message: offset 13).
+ new SeekableStreamEndSequenceNumbers<>(topic, ImmutableMap.of(0, 14L)),
kafkaServer.consumerProperties(),
KafkaSupervisorIOConfig.DEFAULT_POLL_TIMEOUT_MILLIS,
true,
@@ -2594,62 +2590,65 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
kafkaProducer.commitTransaction();
}
- while (countEvents(task) != 2) {
- Thread.sleep(25);
- }
-
+ awaitConsumedOffsets(task, ImmutableMap.of(0, 1L)); // Consume two real messages
Assert.assertEquals(2, countEvents(task));
Assert.assertEquals(Status.READING, task.getRunner().getStatus());
//verify the 2 indexed records
- final QuerySegmentSpec firstInterval = OBJECT_MAPPER.readValue(
- "\"2008/2010\"", QuerySegmentSpec.class
- );
- Iterable<ScanResultValue> scanResultValues = scanData(task, firstInterval);
+ final QuerySegmentSpec firstInterval = OBJECT_MAPPER.readValue("\"2008/2010\"", QuerySegmentSpec.class);
+ Iterable<Map<String, Object>> scanResultValues = scanData(task, firstInterval);
Assert.assertEquals(2, Iterables.size(scanResultValues));
// Insert 3 more records and rollback
try (final KafkaProducer<byte[], byte[]> kafkaProducer = kafkaServer.newProducer()) {
kafkaProducer.initTransactions();
kafkaProducer.beginTransaction();
- for (ProducerRecord<byte[], byte[]> record : Iterables.limit(Iterables.skip(records, 2), 3)) {
+ for (ProducerRecord<byte[], byte[]> record : Iterables.skip(Iterables.limit(records, 5), 2)) {
kafkaProducer.send(record).get();
}
kafkaProducer.flush();
kafkaProducer.abortTransaction();
}
+ // Insert up through first 8 items
+ try (final KafkaProducer<byte[], byte[]> kafkaProducer = kafkaServer.newProducer()) {
+ kafkaProducer.initTransactions();
+ kafkaProducer.beginTransaction();
+ for (ProducerRecord<byte[], byte[]> record : Iterables.skip(Iterables.limit(records, 8), 5)) {
+ kafkaProducer.send(record).get();
+ }
+ kafkaProducer.commitTransaction();
+ }
+
+ awaitConsumedOffsets(task, ImmutableMap.of(0, 9L)); // Consume 8 real messages + 2 txn controls
Assert.assertEquals(2, countEvents(task));
- Assert.assertEquals(Status.READING, task.getRunner().getStatus());
- final QuerySegmentSpec rollbackedInterval = OBJECT_MAPPER.readValue(
- "\"2010/2012\"", QuerySegmentSpec.class
- );
+ final QuerySegmentSpec rollbackedInterval = OBJECT_MAPPER.readValue("\"2010/2012\"", QuerySegmentSpec.class);
scanResultValues = scanData(task, rollbackedInterval);
//verify that there are no records indexed in the rollbacked time period
Assert.assertEquals(0, Iterables.size(scanResultValues));
- // Insert remaining data
+ final QuerySegmentSpec endInterval = OBJECT_MAPPER.readValue("\"2008/2049\"", QuerySegmentSpec.class);
+ Iterable<Map<String, Object>> scanResultValues1 = scanData(task, endInterval);
+ Assert.assertEquals(2, Iterables.size(scanResultValues1));
+
+ // Insert all remaining messages. One will get picked up.
try (final KafkaProducer<byte[], byte[]> kafkaProducer = kafkaServer.newProducer()) {
kafkaProducer.initTransactions();
kafkaProducer.beginTransaction();
- for (ProducerRecord<byte[], byte[]> record : Iterables.skip(records, 5)) {
+ for (ProducerRecord<byte[], byte[]> record : Iterables.skip(records, 8)) {
kafkaProducer.send(record).get();
}
kafkaProducer.commitTransaction();
}
- final QuerySegmentSpec endInterval = OBJECT_MAPPER.readValue(
- "\"2008/2049\"", QuerySegmentSpec.class
- );
- Iterable<ScanResultValue> scanResultValues1 = scanData(task, endInterval);
- Assert.assertEquals(2, Iterables.size(scanResultValues1));
-
+ // Wait for task to exit and publish
Assert.assertEquals(TaskState.SUCCESS, future.get().getStatusCode());
Assert.assertEquals(task.getRunner().getEndOffsets(), task.getRunner().getCurrentOffsets());
// Check metrics
Assert.assertEquals(3, task.getRunner().getRowIngestionMeters().getProcessed());
+ Assert.assertEquals(1, task.getRunner().getRowIngestionMeters().getProcessedWithError());
Assert.assertEquals(3, task.getRunner().getRowIngestionMeters().getUnparseable());
Assert.assertEquals(1, task.getRunner().getRowIngestionMeters().getThrownAway());
@@ -2664,7 +2663,7 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
publishedDescriptors()
);
Assert.assertEquals(
- new KafkaDataSourceMetadata(new SeekableStreamEndSequenceNumbers<>(topic, ImmutableMap.of(0, 13L))),
+ new KafkaDataSourceMetadata(new SeekableStreamEndSequenceNumbers<>(topic, ImmutableMap.of(0, 14L))),
newDataSchemaMetadata()
);
}
@@ -2824,11 +2823,44 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
Assert.assertEquals(task, task1);
}
- private List<ScanResultValue> scanData(final Task task, QuerySegmentSpec spec)
+ /**
+ * Wait for a task to consume certain offsets (inclusive).
+ */
+ private void awaitConsumedOffsets(final KafkaIndexTask task, final Map<Integer, Long> targetOffsets)
+ throws InterruptedException
{
- ScanQuery query = new Druids.ScanQueryBuilder().dataSource(
- NEW_DATA_SCHEMA.getDataSource()).intervals(spec).build();
- return task.getQueryRunner(query).run(QueryPlus.wrap(query)).toList();
+ while (true) {
+ final ConcurrentMap<Integer, Long> currentOffsets = task.getRunner().getCurrentOffsets();
+
+ // For Kafka, currentOffsets are the last read offsets plus one.
+ boolean allDone = true;
+ for (final Map.Entry<Integer, Long> entry : targetOffsets.entrySet()) {
+ final Long currentOffset = currentOffsets.get(entry.getKey());
+ if (currentOffset == null || currentOffset <= entry.getValue()) {
+ allDone = false;
+ break;
+ }
+ }
+
+ if (allDone) {
+ return;
+ } else {
+ Thread.sleep(5);
+ }
+ }
+ }
+
+ private List<Map<String, Object>> scanData(final Task task, QuerySegmentSpec spec)
+ {
+ ScanQuery query = new Druids.ScanQueryBuilder().dataSource(NEW_DATA_SCHEMA.getDataSource())
+ .intervals(spec)
+ .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST)
+ .build();
+
+ return task.getQueryRunner(query)
+ .run(QueryPlus.wrap(query))
+ .flatMap(result -> Sequences.simple((List<Map<String, Object>>) result.getEvents()))
+ .toList();
}
private void insertData() throws ExecutionException, InterruptedException
@@ -2836,7 +2868,8 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
insertData(records);
}
- private void insertData(Iterable<ProducerRecord<byte[], byte[]>> records) throws ExecutionException, InterruptedException
+ private void insertData(Iterable<ProducerRecord<byte[], byte[]>> records)
+ throws ExecutionException, InterruptedException
{
try (final KafkaProducer<byte[], byte[]> kafkaProducer = kafkaServer.newProducer()) {
kafkaProducer.initTransactions();
@@ -2943,28 +2976,28 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
{
return new DefaultQueryRunnerFactoryConglomerate(
ImmutableMap.<Class<? extends Query>, QueryRunnerFactory>builder()
- .put(
- TimeseriesQuery.class,
- new TimeseriesQueryRunnerFactory(
- new TimeseriesQueryQueryToolChest(),
- new TimeseriesQueryEngine(),
- (query, future) -> {
- // do nothing
- }
- )
- )
- .put(
- ScanQuery.class,
- new ScanQueryRunnerFactory(
- new ScanQueryQueryToolChest(
- new ScanQueryConfig(),
- new DefaultGenericQueryMetricsFactory()
- ),
- new ScanQueryEngine(),
- new ScanQueryConfig()
- )
- )
- .build()
+ .put(
+ TimeseriesQuery.class,
+ new TimeseriesQueryRunnerFactory(
+ new TimeseriesQueryQueryToolChest(),
+ new TimeseriesQueryEngine(),
+ (query, future) -> {
+ // do nothing
+ }
+ )
+ )
+ .put(
+ ScanQuery.class,
+ new ScanQueryRunnerFactory(
+ new ScanQueryQueryToolChest(
+ new ScanQueryConfig(),
+ new DefaultGenericQueryMetricsFactory()
+ ),
+ new ScanQueryEngine(),
+ new ScanQueryConfig()
+ )
+ )
+ .build()
);
}
@@ -2972,7 +3005,7 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
{
directory = tempFolder.newFolder();
final TestUtils testUtils = new TestUtils();
- rowIngestionMetersFactory = testUtils.getRowIngestionMetersFactory();
+ RowIngestionMetersFactory rowIngestionMetersFactory = testUtils.getRowIngestionMetersFactory();
final ObjectMapper objectMapper = testUtils.getTestObjectMapper();
for (Module module : new KafkaIndexTaskModule().getJacksonModules()) {
@@ -3135,13 +3168,16 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
//multiple objects in one Kafka record will yield 2 rows in druid
String wellformed = toJsonString(true, "2049", "d2", "y", "10", "22.0", "2.0") +
- toJsonString(true, "2049", "d3", "y", "10", "23.0", "3.0");
+ toJsonString(true, "2049", "d3", "y", "10", "23.0", "3.0");
//multiple objects in one Kafka record but some objects are in ill-formed format
//as a result, the whole ProducerRecord will be discarded
- String illformed = "{\"timestamp\":2049, \"dim1\": \"d4\", \"dim2\":\"x\", \"dimLong\": 10, \"dimFloat\":\"24.0\", \"met1\":\"2.0\" }" +
- "{\"timestamp\":2049, \"dim1\": \"d5\", \"dim2\":\"y\", \"dimLong\": 10, \"dimFloat\":\"24.0\", \"met1\":invalidFormat }" +
- "{\"timestamp\":2049, \"dim1\": \"d6\", \"dim2\":\"z\", \"dimLong\": 10, \"dimFloat\":\"24.0\", \"met1\":\"3.0\" }";
+ String malformed =
+ "{\"timestamp\":2049, \"dim1\": \"d4\", \"dim2\":\"x\", \"dimLong\": 10, \"dimFloat\":\"24.0\", \"met1\":\"2.0\" }"
+ +
+ "{\"timestamp\":2049, \"dim1\": \"d5\", \"dim2\":\"y\", \"dimLong\": 10, \"dimFloat\":\"24.0\", \"met1\":invalidFormat }"
+ +
+ "{\"timestamp\":2049, \"dim1\": \"d6\", \"dim2\":\"z\", \"dimLong\": 10, \"dimFloat\":\"24.0\", \"met1\":\"3.0\" }";
ProducerRecord<byte[], byte[]>[] producerRecords = new ProducerRecord[]{
// pretty formatted
@@ -3149,7 +3185,7 @@ public class KafkaIndexTaskTest extends SeekableStreamIndexTaskTestBase
//well-formed
new ProducerRecord<>(topic, 0, null, StringUtils.toUtf8(wellformed)),
//ill-formed
- new ProducerRecord<>(topic, 0, null, StringUtils.toUtf8(illformed)),
+ new ProducerRecord<>(topic, 0, null, StringUtils.toUtf8(malformed)),
//a well-formed record after ill-formed to demonstrate that the ill-formed can be successfully skipped
new ProducerRecord<>(topic, 0, null, jbb(true, "2049", "d7", "y", "10", "20.0", "1.0"))
};
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTestBase.java b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTestBase.java
index 6078121e98..d1bc388752 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTestBase.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTestBase.java
@@ -348,7 +348,7 @@ public class SeekableStreamIndexTaskTestBase extends EasyMockSupport
List<SegmentDescriptor> actualDescriptors
) throws IOException
{
- Assert.assertEquals(expectedDescriptors.size(), actualDescriptors.size());
+ Assert.assertEquals("number of segments", expectedDescriptors.size(), actualDescriptors.size());
final Comparator<SegmentDescriptor> comparator = (s1, s2) -> {
final int intervalCompare = Comparators.intervalsByStartThenEnd().compare(s1.getInterval(), s2.getInterval());
if (intervalCompare == 0) {
@@ -379,7 +379,9 @@ public class SeekableStreamIndexTaskTestBase extends EasyMockSupport
if (expectedDesc.expectedDim1Values.isEmpty()) {
continue; // Treating empty expectedDim1Values as a signal that checking the dim1 column value is not needed.
}
- Assertions.assertThat(readSegmentColumn("dim1", actualDesc)).isIn(expectedDesc.expectedDim1Values);
+ Assertions.assertThat(readSegmentColumn("dim1", actualDesc))
+ .describedAs("dim1 values")
+ .isIn(expectedDesc.expectedDim1Values);
}
}
@@ -447,7 +449,7 @@ public class SeekableStreamIndexTaskTestBase extends EasyMockSupport
new LongSumAggregatorFactory("rows", "rows")
)
).granularity(Granularities.ALL)
- .intervals("0000/3000")
+ .intervals(Intervals.ONLY_ETERNITY)
.build();
List<Result<TimeseriesResultValue>> results = task.getQueryRunner(query).run(QueryPlus.wrap(query)).toList();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org