You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by jo...@apache.org on 2019/03/15 00:12:04 UTC
[incubator-druid] branch 0.14.0-incubating updated: Fix
KafkaRecordSupplier assign (#7260) (#7265)
This is an automated email from the ASF dual-hosted git repository.
jonwei pushed a commit to branch 0.14.0-incubating
in repository https://gitbox.apache.org/repos/asf/incubator-druid.git
The following commit(s) were added to refs/heads/0.14.0-incubating by this push:
new acdda58 Fix KafkaRecordSupplier assign (#7260) (#7265)
acdda58 is described below
commit acdda58d232668d06cb0e32ef08d20f9c24fa284
Author: Clint Wylie <cj...@gmail.com>
AuthorDate: Thu Mar 14 17:11:57 2019 -0700
Fix KafkaRecordSupplier assign (#7260) (#7265)
* Fix KafkaRecordSupplier assign
* TeamCity fix
---
.../druid/indexing/kafka/KafkaRecordSupplier.java | 1 -
.../druid/indexing/kafka/KafkaIndexTaskTest.java | 59 ++++++++++++++++++++++
2 files changed, 59 insertions(+), 1 deletion(-)
diff --git a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaRecordSupplier.java b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaRecordSupplier.java
index 935404c..60aea3c 100644
--- a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaRecordSupplier.java
+++ b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaRecordSupplier.java
@@ -68,7 +68,6 @@ public class KafkaRecordSupplier implements RecordSupplier<Integer, Long>
.stream()
.map(x -> new TopicPartition(x.getStream(), x.getPartitionId()))
.collect(Collectors.toSet()));
- seekToEarliest(streamPartitions);
}
@Override
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 a2224e5..d2f7373 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
@@ -2295,6 +2295,65 @@ public class KafkaIndexTaskTest
Assert.assertEquals(Status.READING, task.getRunner().getStatus());
}
+ @Test(timeout = 60_000L)
+ public void testCanStartFromLaterThanEarliestOffset() throws Exception
+ {
+ if (!isIncrementalHandoffSupported) {
+ return;
+ }
+ final String baseSequenceName = "sequence0";
+ maxRowsPerSegment = Integer.MAX_VALUE;
+ maxTotalRows = null;
+
+ // Insert data
+ try (final KafkaProducer<byte[], byte[]> kafkaProducer = kafkaServer.newProducer()) {
+ for (ProducerRecord<byte[], byte[]> record : records) {
+ kafkaProducer.send(record).get();
+ }
+ }
+
+ Map<String, Object> consumerProps = kafkaServer.consumerProperties();
+ consumerProps.put("max.poll.records", "1");
+
+ final SeekableStreamPartitions<Integer, Long> startPartitions = new SeekableStreamPartitions<>(
+ topic,
+ ImmutableMap.of(
+ 0,
+ 0L,
+ 1,
+ 1L
+ )
+ );
+
+ final SeekableStreamPartitions<Integer, Long> endPartitions = new SeekableStreamPartitions<>(
+ topic,
+ ImmutableMap.of(
+ 0,
+ 10L,
+ 1,
+ 2L
+ )
+ );
+
+ final KafkaIndexTask task = createTask(
+ null,
+ new KafkaIndexTaskIOConfig(
+ 0,
+ baseSequenceName,
+ startPartitions,
+ endPartitions,
+ consumerProps,
+ KafkaSupervisorIOConfig.DEFAULT_POLL_TIMEOUT_MILLIS,
+ true,
+ null,
+ null,
+ false
+ )
+ );
+ final ListenableFuture<TaskStatus> future = runTask(task);
+ Assert.assertEquals(TaskState.SUCCESS, future.get().getStatusCode());
+ }
+
private ListenableFuture<TaskStatus> runTask(final Task task)
{
try {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org