Support kafka transactional topics

docs/content/development/extensions-core/kafka-ingestion.md

@@ -201,7 +201,6 @@ For Roaring bitmaps:
 |`completionTimeout`|ISO8601 Period|The length of time to wait before declaring a publishing task as failed and terminating it. If this is set too low, your tasks may never publish. The publishing clock for a task begins roughly after `taskDuration` elapses.|no (default == PT30M)|
 |`lateMessageRejectionPeriod`|ISO8601 Period|Configure tasks to reject messages with timestamps earlier than this period before the task was created; for example if this is set to `PT1H` and the supervisor creates a task at *2016-01-01T12:00Z*, messages with timestamps earlier than *2016-01-01T11:00Z* will be dropped. This may help prevent concurrency issues if your data stream has late messages and you have multiple pipelines that need to operate on the same segments (e.g. a realtime and a nightly batch ingestion pipeline).|no (default == none)|
 |`earlyMessageRejectionPeriod`|ISO8601 Period|Configure tasks to reject messages with timestamps later than this period after the task reached its taskDuration; for example if this is set to `PT1H`, the taskDuration is set to `PT1H` and the supervisor creates a task at *2016-01-01T12:00Z*, messages with timestamps later than *2016-01-01T14:00Z* will be dropped. **Note:** Tasks sometimes run past their task duration, for example, in cases of supervisor failover. Setting earlyMessageRejectionPeriod too low may cause messages to be dropped unexpectedly whenever a task runs past its originally configured task duration.|no (default == none)|
-|`skipOffsetGaps`|Boolean|Whether or not to allow gaps of missing offsets in the Kafka stream. This is required for compatibility with implementations such as MapR Streams which does not guarantee consecutive offsets. If this is false, an exception will be thrown if offsets are not consecutive.|no (default == false)|
 
 ## Operations
 

docs/content/development/extensions.md

@@ -48,7 +48,7 @@ Core extensions are maintained by Druid committers.
 |druid-datasketches|Support for approximate counts and set operations with [DataSketches](http://datasketches.github.io/).|[link](../development/extensions-core/datasketches-extension.html)|
 |druid-hdfs-storage|HDFS deep storage.|[link](../development/extensions-core/hdfs.html)|
 |druid-histogram|Approximate histograms and quantiles aggregator.|[link](../development/extensions-core/approximate-histograms.html)|
-|druid-kafka-eight|Kafka ingest firehose (high level consumer) for realtime nodes.|[link](../development/extensions-core/kafka-eight-firehose.html)|
+|druid-kafka-eight|Kafka ingest firehose (high level consumer) for realtime nodes(deprecated).|[link](../development/extensions-core/kafka-eight-firehose.html)|
 |druid-kafka-extraction-namespace|Kafka-based namespaced lookup. Requires namespace lookup extension.|[link](../development/extensions-core/kafka-extraction-namespace.html)|
 |druid-kafka-indexing-service|Supervised exactly-once Kafka ingestion for the indexing service.|[link](../development/extensions-core/kafka-ingestion.html)|
 |druid-kinesis-indexing-service|Supervised exactly-once Kinesis ingestion for the indexing service.|[link](../development/extensions-core/kinesis-ingestion.html)|
@@ -81,7 +81,7 @@ All of these community extensions can be downloaded using *pull-deps* with the c
 |druid-cassandra-storage|Apache Cassandra deep storage.|[link](../development/extensions-contrib/cassandra.html)|
 |druid-cloudfiles-extensions|Rackspace Cloudfiles deep storage and firehose.|[link](../development/extensions-contrib/cloudfiles.html)|
 |druid-distinctcount|DistinctCount aggregator|[link](../development/extensions-contrib/distinctcount.html)|
-|druid-kafka-eight-simpleConsumer|Kafka ingest firehose (low level consumer).|[link](../development/extensions-contrib/kafka-simple.html)|
+|druid-kafka-eight-simpleConsumer|Kafka ingest firehose (low level consumer)(deprecated).|[link](../development/extensions-contrib/kafka-simple.html)|
 |druid-orc-extensions|Support for data in Apache Orc data format.|[link](../development/extensions-contrib/orc.html)|
 |druid-rabbitmq|RabbitMQ firehose.|[link](../development/extensions-contrib/rabbitmq.html)|
 |druid-redis-cache|A cache implementation for Druid based on Redis.|[link](../development/extensions-contrib/redis-cache.html)|

extensions-contrib/kafka-eight-simpleConsumer/src/main/java/org/apache/druid/firehose/kafka/KafkaEightSimpleConsumerDruidModule.java

@@ -28,6 +28,7 @@
 
 import java.util.List;
 
+@Deprecated
 public class KafkaEightSimpleConsumerDruidModule implements DruidModule
 {
   @Override

extensions-contrib/kafka-eight-simpleConsumer/src/main/java/org/apache/druid/firehose/kafka/KafkaEightSimpleConsumerFirehoseFactory.java

@@ -46,6 +46,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+@Deprecated
 public class KafkaEightSimpleConsumerFirehoseFactory implements
     FirehoseFactoryV2<ByteBufferInputRowParser>
 {

extensions-contrib/kafka-eight-simpleConsumer/src/main/java/org/apache/druid/firehose/kafka/KafkaSimpleConsumer.java

@@ -56,6 +56,7 @@
  * This class is not thread safe, the caller must ensure all the methods be
  * called from single thread
  */
+@Deprecated
 public class KafkaSimpleConsumer
 {
 

extensions-core/kafka-eight/src/main/java/org/apache/druid/firehose/kafka/KafkaEightDruidModule.java

@@ -30,6 +30,7 @@
 
 /**
  */
+@Deprecated
 public class KafkaEightDruidModule implements DruidModule
 {
   @Override

extensions-core/kafka-eight/src/main/java/org/apache/druid/firehose/kafka/KafkaEightFirehoseFactory.java

@@ -47,7 +47,9 @@
 import java.util.Set;
 
 /**
+ * This class is deprecated and kafka-eight module should be removed completely
  */
+@Deprecated
 public class KafkaEightFirehoseFactory implements FirehoseFactory<InputRowParser<ByteBuffer>>
 {
   private static final Logger log = new Logger(KafkaEightFirehoseFactory.class);

extensions-core/kafka-indexing-service/pom.xml

@@ -34,7 +34,7 @@
   </parent>
 
   <properties>
-    <apache.kafka.version>0.10.2.2</apache.kafka.version>
+    <apache.kafka.version>2.1.0</apache.kafka.version>
   </properties>
 
   <dependencies>

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java

@@ -108,6 +108,7 @@ KafkaConsumer<byte[], byte[]> newConsumer()
       props.setProperty("auto.offset.reset", "none");
       props.setProperty("key.deserializer", ByteArrayDeserializer.class.getName());
       props.setProperty("value.deserializer", ByteArrayDeserializer.class.getName());
+      props.setProperty("isolation.level", "read_committed");
 
       return new KafkaConsumer<>(props);
     }

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskIOConfig.java

@@ -45,8 +45,7 @@ public KafkaIndexTaskIOConfig(
       @JsonProperty("pollTimeout") Long pollTimeout,
       @JsonProperty("useTransaction") Boolean useTransaction,
       @JsonProperty("minimumMessageTime") DateTime minimumMessageTime,
-      @JsonProperty("maximumMessageTime") DateTime maximumMessageTime,
-      @JsonProperty("skipOffsetGaps") Boolean skipOffsetGaps
+      @JsonProperty("maximumMessageTime") DateTime maximumMessageTime
   )
   {
     super(
@@ -57,7 +56,6 @@ public KafkaIndexTaskIOConfig(
         useTransaction,
         minimumMessageTime,
         maximumMessageTime,
-        skipOffsetGaps,
         null
     );
 
@@ -100,7 +98,6 @@ public String toString()
            ", useTransaction=" + isUseTransaction() +
            ", minimumMessageTime=" + getMinimumMessageTime() +
            ", maximumMessageTime=" + getMaximumMessageTime() +
-           ", skipOffsetGaps=" + isSkipOffsetGaps() +
            '}';
   }
 }

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaRecordSupplier.java

@@ -36,6 +36,7 @@
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 
 import javax.annotation.Nonnull;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -110,7 +111,7 @@ public Set<StreamPartition<Integer>> getAssignment()
   public List<OrderedPartitionableRecord<Integer, Long>> poll(long timeout)
   {
     List<OrderedPartitionableRecord<Integer, Long>> polledRecords = new ArrayList<>();
-    for (ConsumerRecord<byte[], byte[]> record : consumer.poll(timeout)) {
+    for (ConsumerRecord<byte[], byte[]> record : consumer.poll(Duration.ofMillis(timeout))) {
       polledRecords.add(new OrderedPartitionableRecord<>(
           record.topic(),
           record.partition(),

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/LegacyKafkaIndexTaskRunner.java

@@ -413,23 +413,6 @@ public void run()
             }
 
             if (record.offset() < endOffsets.get(record.partition())) {
-              if (record.offset() != nextOffsets.get(record.partition())) {
-                if (ioConfig.isSkipOffsetGaps()) {
-                  log.warn(
-                      "Skipped to offset[%,d] after offset[%,d] in partition[%d].",
-                      record.offset(),
-                      nextOffsets.get(record.partition()),
-                      record.partition()
-                  );
-                } else {
-                  throw new ISE(
-                      "WTF?! Got offset[%,d] after offset[%,d] in partition[%d].",
-                      record.offset(),
-                      nextOffsets.get(record.partition()),
-                      record.partition()
-                  );
-                }
-              }
 
               try {
                 final byte[] valueBytes = record.value();
@@ -489,7 +472,7 @@ public void run()
               nextOffsets.put(record.partition(), record.offset() + 1);
             }
 
-            if (nextOffsets.get(record.partition()).equals(endOffsets.get(record.partition()))
+            if (nextOffsets.get(record.partition()) >= (endOffsets.get(record.partition()))
                 && assignment.remove(record.partition())) {
               log.info("Finished reading topic[%s], partition[%,d].", record.topic(), record.partition());
               KafkaIndexTask.assignPartitions(consumer, topic, assignment);

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java

@@ -212,8 +212,7 @@ protected SeekableStreamIndexTaskIOConfig createTaskIoConfig(
         kafkaIoConfig.getPollTimeout(),
         true,
         minimumMessageTime,
-        maximumMessageTime,
-        kafkaIoConfig.isSkipOffsetGaps()
+        maximumMessageTime
     );
   }
 

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfig.java

@@ -38,7 +38,7 @@ public class KafkaSupervisorIOConfig extends SeekableStreamSupervisorIOConfig
 
   private final Map<String, Object> consumerProperties;
   private final long pollTimeout;
-  private final boolean skipOffsetGaps;
+
 
   @JsonCreator
   public KafkaSupervisorIOConfig(
@@ -53,8 +53,7 @@ public KafkaSupervisorIOConfig(
       @JsonProperty("useEarliestOffset") Boolean useEarliestOffset,
       @JsonProperty("completionTimeout") Period completionTimeout,
       @JsonProperty("lateMessageRejectionPeriod") Period lateMessageRejectionPeriod,
-      @JsonProperty("earlyMessageRejectionPeriod") Period earlyMessageRejectionPeriod,
-      @JsonProperty("skipOffsetGaps") Boolean skipOffsetGaps
+      @JsonProperty("earlyMessageRejectionPeriod") Period earlyMessageRejectionPeriod
   )
   {
     super(
@@ -76,7 +75,6 @@ public KafkaSupervisorIOConfig(
         StringUtils.format("consumerProperties must contain entry for [%s]", BOOTSTRAP_SERVERS_KEY)
     );
     this.pollTimeout = pollTimeout != null ? pollTimeout : DEFAULT_POLL_TIMEOUT_MILLIS;
-    this.skipOffsetGaps = skipOffsetGaps != null ? skipOffsetGaps : false;
   }
 
   @JsonProperty
@@ -103,12 +101,6 @@ public boolean isUseEarliestOffset()
     return isUseEarliestSequenceNumber();
   }
 
-  @JsonProperty
-  public boolean isSkipOffsetGaps()
-  {
-    return skipOffsetGaps;
-  }
-
   @Override
   public String toString()
   {
@@ -125,7 +117,6 @@ public String toString()
            ", completionTimeout=" + getCompletionTimeout() +
            ", earlyMessageRejectionPeriod=" + getEarlyMessageRejectionPeriod() +
            ", lateMessageRejectionPeriod=" + getLateMessageRejectionPeriod() +
-           ", skipOffsetGaps=" + skipOffsetGaps +
            '}';
   }
 

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIOConfigTest.java

@@ -77,7 +77,6 @@ public void testSerdeWithDefaults() throws Exception
     Assert.assertTrue(config.isUseTransaction());
     Assert.assertFalse("minimumMessageTime", config.getMinimumMessageTime().isPresent());
     Assert.assertFalse("maximumMessageTime", config.getMaximumMessageTime().isPresent());
-    Assert.assertFalse("skipOffsetGaps", config.isSkipOffsetGaps());
     Assert.assertEquals(Collections.EMPTY_SET, config.getExclusiveStartSequenceNumberPartitions());
   }
 
@@ -93,8 +92,7 @@ public void testSerdeWithNonDefaults() throws Exception
                      + "  \"consumerProperties\": {\"bootstrap.servers\":\"localhost:9092\"},\n"
                      + "  \"useTransaction\": false,\n"
                      + "  \"minimumMessageTime\": \"2016-05-31T12:00Z\",\n"
-                     + "  \"maximumMessageTime\": \"2016-05-31T14:00Z\",\n"
-                     + "  \"skipOffsetGaps\": true\n"
+                     + "  \"maximumMessageTime\": \"2016-05-31T14:00Z\"\n"
                      + "}";
 
     KafkaIndexTaskIOConfig config = (KafkaIndexTaskIOConfig) mapper.readValue(
@@ -115,9 +113,7 @@ public void testSerdeWithNonDefaults() throws Exception
     Assert.assertFalse(config.isUseTransaction());
     Assert.assertEquals(DateTimes.of("2016-05-31T12:00Z"), config.getMinimumMessageTime().get());
     Assert.assertEquals(DateTimes.of("2016-05-31T14:00Z"), config.getMaximumMessageTime().get());
-    Assert.assertTrue("skipOffsetGaps", config.isSkipOffsetGaps());
     Assert.assertEquals(Collections.EMPTY_SET, config.getExclusiveStartSequenceNumberPartitions());
-
   }
 
   @Test