[jira] [Created] (GOBBLIN-393) Invalid Watermark in kafka Streaming stops kafka ingestion

["ROHIT KALHANS (JIRA)" <ji...@apache.org>]

ROHIT KALHANS created GOBBLIN-393:
-------------------------------------

             Summary: Invalid Watermark in kafka Streaming stops kafka ingestion
                 Key: GOBBLIN-393
                 URL: https://issues.apache.org/jira/browse/GOBBLIN-393
             Project: Apache Gobblin
          Issue Type: Bug
         Environment: OS: Redhat 
version : gobblin 0.13
Kafka version: Kafka 1.0.0
            Reporter: ROHIT KALHANS


When using Kafka to Kafka streaming, I keep hitting this error post which the ingestion streaming stops.

 
{noformat}
ERROR [22:54:35.306] [kafka-producer-network-thread | gobblin] o.a.k.c.u.KafkaThread [KafkaThread.java:30] - Uncaught exception in kafka-producer-network-thread | gobblin: java.lang.AssertionError: The acknowledgement counter for this watermark went negative. Please file a bug! at org.apache.gobblin.writer.AcknowledgableWatermark.ack(AcknowledgableWatermark.java:42) at org.apache.gobblin.stream.StreamEntity.ack(StreamEntity.java:82) at org.apache.gobblin.writer.AsyncWriterManager$1.onSuccess(AsyncWriterManager.java:321) at org.apache.gobblin.writer.AsyncWriterManager$1.onSuccess(AsyncWriterManager.java:316) at org.apache.gobblin.kafka.writer.Kafka09DataWriter$2.onCompletion(Kafka09DataWriter.java:124) at org.apache.kafka.clients.producer.internals.RecordBatch.done(RecordBatch.java:97) at org.apache.kafka.clients.producer.internals.Sender.completeBatch(Sender.java:299) at org.apache.kafka.clients.producer.internals.Sender.handleProduceResponse(Sender.java:260) at org.apache.kafka.clients.producer.internals.Sender.access$100(Sender.java:56) at org.apache.kafka.clients.producer.internals.Sender$1.onComplete(Sender.java:342) at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:274) at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:216) at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:128) at java.lang.Thread.run(Thread.java:748) {noformat}
Here are the key points:
 # Gobblin is running in the embedded mode. It is a part of a bigger application.
 #  here is the config for the job in Json format. the json is converted to properties internally

{noformat}
"config":{ "job.lock.enabled": "false", "task.executionMode": "STREAMING", "gobblin.streaming.kafka.topic.key.deserializer": "org.apache.kafka.common.serialization.StringDeserializer", "gobblin.streaming.kafka.topic.value.deserializer": "org.apache.kafka.common.serialization.ByteArrayDeserializer", "source.class": "org.apache.gobblin.source.extractor.extract.kafka.KafkaSimpleStreamingSource", "gobblin.streaming.kafka.topic.singleton": "archive", "kafka.brokers": "localhost:9092", "streaming.watermarkStateStore.type": "mysql", "state.store.db.url": "jdbc:mysql://localhost:3306/Saboo", "state.store.db.user": "saboo_root", "state.store.db.password": "sabootest", "streaming.watermarkStateStore.config.state.store.zk.connectString": "localhost:2181", "streaming.watermark.commitIntervalMillis": "2000", "converter.classes": "org.apache.gobblin.converter.SamplingConverter", "converter.sample.ratio": "0.10", "writer.builder.class": "org.apache.gobblin.kafka.writer.KafkaDataWriterBuilder", "writer.kafka.topic": "archive2", "writer.kafka.producerConfig.bootstrap.servers": "localhost:9092", "writer.kafka.producerConfig.value.serializer": "org.apache.kafka.common.serialization.ByteArraySerializer", "data.publisher.type": "org.apache.gobblin.publisher.NoopPublisher" }{noformat}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)