sergioferragut opened a new issue, #14041: URL: https://github.com/apache/druid/issues/14041
### Affected Version 25.0.0 ### Description This came up in a thread in slack: https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680631762258829 With streaming ingestion that uses: ``` "maxParseExceptions": 2147483647, "maxSavedParseExceptions": 0 ``` The expectation is that it will not fail when it hits parsing errors. But if a byteEntityReader is specified as in avroBtesDecoder in this example: ``` "inputFormat": { "type": "kafka", "valueFormat": { "type": "avro_stream", "avroBytesDecoder": { "type": "glo_schema" #this is our custom decoder, where we process schema that is embedded in the message, and throws ParseException based on schema/payload validation } }, ``` Looking at the code, it seems like the problem appears when a byteEntityReader is specified for streaming ingestion which makes use of [StreamChunkParser.parseWithInputFormat]([https://github.com/apache/druid/blob/58a3acc2c4c5d5d22098e15f7ff25d6e9baa8541/inde[…]org/apache/druid/indexing/seekablestream/StreamChunkParser.java](https://github.com/apache/druid/blob/58a3acc2c4c5d5d22098e15f7ff25d6e9baa8541/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/StreamChunkParser.java#L148)). Seems like ParseExceptions thrown by the call to `byteEntityReader.read()` will not hit the parseExceptionHandler causing it to fail instead. Not sure what the protocol is here when referencing a slack thread, but I'm pasting it all here: ``` Vamsi Gandrapu [2 days ago](https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680631762258829) Hi, we are using Kafka ingestion with avro-stream. We expect decoder to result in ParseException for some messages. Below config is applied on the ingestion task to tolerate parse errors and skip messages. However, it seems like the tasks fail and supervisor status changes to unhealthy . Does not look like consumer is progressing to process messages that are expected to be good. "tuningConfig": { "type": "kafka", "maxParseExceptions": 2147483647, "maxSavedParseExceptions": 0 } 11 replies Sergio Ferragut [1 day ago](https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680650196593099?thread_ts=1680631762.258829&cid=C0309C9L90D) Can you share a task log with the exception. What Druid version are you on? Vamsi Gandrapu [1 day ago](https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680656329599159?thread_ts=1680631762.258829&cid=C0309C9L90D) We are on v25.0.0 logs here 2023-04-05T00:55:23,310 INFO [task-runner-0-priority-0] org.apache.druid.indexing.worker.executor.ExecutorLifecycle - Task completed with status: { "id" : "index_kafka_private.DATAPLATFORMV2.events_3f76be529bcb647_gcgcihkh", "status" : "FAILED", "duration" : 3397, "errorMsg" : "org.apache.druid.java.util.common.parsers.ParseException: Failed to parse avro message.\n\tat org.apac...", "location" : { "host" : null, "port" : -1, "tlsPort" : -1 } } 2023-04-05T00:55:23,315 INFO [main] org.apache.druid.java.util.common.lifecycle.Lifecycle - Stopping lifecycle [module] stage [ANNOUNCEMENTS] 2023-04-05T00:55:23,327 INFO [main] org.apache.druid.java.util.common.lifecycle.Lifecycle - Stopping lifecycle [module] stage [SERVER] 2023-04-05T00:55:23,331 INFO [main] org.eclipse.jetty.server.AbstractConnector - Stopped ServerConnector@2e60d4f4{HTTP/1.1, (http/1.1)}{0.0.0.0:8100} 2023-04-05T00:55:23,332 INFO [main] org.eclipse.jetty.server.session - node0 Stopped scavenging 2023-04-05T00:55:23,333 INFO [main] org.eclipse.jetty.server.handler.ContextHandler - Stopped o.e.j.s.ServletContextHandler@272485a6{/,null,STOPPED} 2023-04-05T00:55:23,336 INFO [main] org.apache.druid.java.util.common.lifecycle.Lifecycle - Stopping lifecycle [module] stage [NORMAL] 2023-04-05T00:55:23,336 INFO [main] org.apache.druid.server.coordination.ZkCoordinator - Stopping ZkCoordinator for [DruidServerMetadata{name='10.40.89.93:8100', hostAndPort='10.40.89.93:8100', hostAndTlsPort='null', maxSize=0, tier='_default_tier', type=indexer-executor, priority=0}] 2023-04-05T00:55:23,336 INFO [main] org.apache.druid.server.coordination.SegmentLoadDropHandler - Stopping... 2023-04-05T00:55:23,337 INFO [main] org.apache.druid.server.coordination.SegmentLoadDropHandler - Stopped. 2023-04-05T00:55:23,337 INFO [main] org.apache.druid.indexing.overlord.SingleTaskBackgroundRunner - Starting graceful shutdown of task[index_kafka_private.DATAPLATFORMV2.events_3f76be529bcb647_gcgcihkh]. 2023-04-05T00:55:23,337 INFO [main] org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner - Stopping forcefully (status: [READING]) 2023-04-05T00:55:23,406 INFO [LookupExtractorFactoryContainerProvider-MainThread] org.apache.druid.query.lookup.LookupReferencesManager - Lookup Management loop exited. Lookup notices are not handled anymore. 2023-04-05T00:55:23,407 INFO [main] org.apache.druid.security.basic.authorization.db.cache.CoordinatorPollingBasicAuthorizerCacheManager - CoordinatorPollingBasicAuthorizerCacheManager is stopping. 2023-04-05T00:55:23,407 INFO [main] org.apache.druid.security.basic.authorization.db.cache.CoordinatorPollingBasicAuthorizerCacheManager - CoordinatorPollingBasicAuthorizerCacheManager is stopped. 2023-04-05T00:55:23,407 INFO [main] org.apache.druid.security.basic.authentication.db.cache.CoordinatorPollingBasicAuthenticatorCacheManager - CoordinatorPollingBasicAuthenticatorCacheManager is stopping. 2023-04-05T00:55:23,407 INFO [main] org.apache.druid.security.basic.authentication.db.cache.CoordinatorPollingBasicAuthenticatorCacheManager - CoordinatorPollingBasicAuthenticatorCacheManager is stopped. 2023-04-05T00:55:23,408 INFO [Curator-Framework-0] org.apache.curator.framework.imps.CuratorFrameworkImpl - backgroundOperationsLoop exiting 2023-04-05T00:55:23,513 INFO [main-EventThread] org.apache.zookeeper.ClientCnxn - EventThread shut down for session: 0x10005abdb2e053b 2023-04-05T00:55:23,513 INFO [main] org.apache.zookeeper.ZooKeeper - Session: 0x10005abdb2e053b closed 2023-04-05T00:55:23,517 INFO [main] org.apache.druid.java.util.common.lifecycle.Lifecycle - Stopping lifecycle [module] stage [INIT] Finished peon task Vamsi Gandrapu [1 day ago](https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680709606857679?thread_ts=1680631762.258829&cid=C0309C9L90D) [@Sergio Ferragut](https://apachedruidworkspace.slack.com/team/U030K4UM3H7) fyi ^^ Sergio Ferragut [1 day ago](https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680709919399539?thread_ts=1680631762.258829&cid=C0309C9L90D) I'm looking into it. I have the same expectation you do. It should skip the parse errors, I need to spend some time in the code to try to understand where it might be going wrong. Can you share your ingestion spec? Vamsi Gandrapu [23 hours ago](https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680724271262929?thread_ts=1680631762.258829&cid=C0309C9L90D) here is the spec., pretty much standard except that we use a custom avro decoder that we wrote to handle avro schema embedded in the message rather than inline or registry based. { "type": "kafka", "spec": { "ioConfig": { "type": "kafka", "consumerProperties": { .... }, "topic": "topic-abc", "inputFormat": { "type": "kafka", "valueFormat": { "type": "avro_stream", "avroBytesDecoder": { "type": "glo_schema" #this is our custom decoder, where we process schema that is embedded in the message, and throws ParseException based on schema/payload validation } }, "flattenSpec": { "useFieldDiscovery": true }, "binaryAsString": false }, "appendToExisting": false, "useEarliestOffset": true, "idleConfig": { "enabled": true, "inactiveAfterMillis": 300000 } }, "tuningConfig": { "type": "kafka", "maxParseExceptions": 2147483647, "maxSavedParseExceptions": 0 }, "dataSchema": { ..... } } } Sergio Ferragut [21 hours ago](https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680730544948679?thread_ts=1680631762.258829&cid=C0309C9L90D) Here's what I found. I'm not sure this is the reason, I think a committer may need to take a look: [https://github.com/apache/druid/blob/58a3acc2c4c5d5d22098e15f7ff25d6e9baa8541/inde[…]org/apache/druid/indexing/seekablestream/StreamChunkParser.java](https://github.com/apache/druid/blob/58a3acc2c4c5d5d22098e15f7ff25d6e9baa8541/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/StreamChunkParser.java#L148) I'm wondering whether the byteEntityReader.read() is referencing you reader and when/if it throws the ParseException here it would not use the ParseExceptionHandler. Not sure what the solution is, but I think this may be the cause. [StreamChunkParser.java](https://github.com/apache/druid/blob/58a3acc2c4c5d5d22098e15f7ff25d6e9baa8541/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/StreamChunkParser.java) byteEntityReader.read(), <https://github.com/[apache/druid](https://github.com/apache/druid)|apache/druid>apache/druid | Added by [GitHub](https://apachedruidworkspace.slack.com/services/B04LTM2RGJD) Sergio Ferragut [21 hours ago](https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680730740621429?thread_ts=1680631762.258829&cid=C0309C9L90D) The resulting iterator has the parseException logic in it here: [https://github.com/apache/druid/blob/f2ac6cd96edfacca6ba7880d5eb9c532ea0b95f2/inde[…]id/indexing/common/task/FilteringCloseableInputRowIterator.java](https://github.com/apache/druid/blob/f2ac6cd96edfacca6ba7880d5eb9c532ea0b95f2/indexing-service/src/main/java/org/apache/druid/indexing/common/task/FilteringCloseableInputRowIterator.java#L60) But it won't get there if the byteEntityReader.read() throws the exception. [FilteringCloseableInputRowIterator.java](https://github.com/apache/druid/blob/f2ac6cd96edfacca6ba7880d5eb9c532ea0b95f2/indexing-service/src/main/java/org/apache/druid/indexing/common/task/FilteringCloseableInputRowIterator.java) public boolean hasNext() <https://github.com/[apache/druid](https://github.com/apache/druid)|apache/druid>apache/druid | Added by [GitHub](https://apachedruidworkspace.slack.com/services/B04LTM2RGJD) Sergio Ferragut [21 hours ago](https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680730783693369?thread_ts=1680631762.258829&cid=C0309C9L90D) A more detailed stack trace from the full task log would help us confirm. Vamsi Gandrapu [14 hours ago](https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680756290882979?thread_ts=1680631762.258829&cid=C0309C9L90D) [@Sergio Ferragut](https://apachedruidworkspace.slack.com/team/U030K4UM3H7) as you mentioned, apparently looks like its not hitting the handler. 2023-04-06T04:38:07,887 ERROR [task-runner-0-priority-0] org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner - Encountered exception in run() before persisting. org.apache.druid.java.util.common.parsers.ParseException: Failed to parse avro message from this origin:unknown at org.apache.druid.data.input.avro.GlobalityAvroBytesDecoder.parse(GlobalityAvroBytesDecoder.java:111) ~[?:?] at org.apache.druid.data.input.avro.AvroStreamReader.intermediateRowIterator(AvroStreamReader.java:70) ~[?:?] at org.apache.druid.data.input.IntermediateRowParsingReader.intermediateRowIteratorWithMetadata(IntermediateRowParsingReader.java:231) ~[druid-core-25.0.0.jar:25.0.0] at org.apache.druid.data.input.IntermediateRowParsingReader.read(IntermediateRowParsingReader.java:49) ~[druid-core-25.0.0.jar:25.0.0] at org.apache.druid.data.input.kafkainput.KafkaInputReader.buildBlendedRows(KafkaInputReader.java:158) ~[?:?] at org.apache.druid.data.input.kafkainput.KafkaInputReader.read(KafkaInputReader.java:129) ~[?:?] at org.apache.druid.segment.transform.TransformingInputEntityReader.read(TransformingInputEntityReader.java:43) ~[druid-processing-25.0.0.jar:25.0.0] at org.apache.druid.indexing.seekablestream.SettableByteEntityReader.read(SettableByteEntityReader.java:70) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.seekablestream.StreamChunkParser.parseWithInputFormat(StreamChunkParser.java:135) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.seekablestream.StreamChunkParser.parse(StreamChunkParser.java:104) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.runInternal(SeekableStreamIndexTaskRunner.java:635) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.run(SeekableStreamIndexTaskRunner.java:266) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTask.runTask(SeekableStreamIndexTask.java:151) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.common.task.AbstractTask.run(AbstractTask.java:169) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.overlord.SingleTaskBackgroundRunner$SingleTaskBackgroundRunnerCallable.call(SingleTaskBackgroundRunner.java:477) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.overlord.SingleTaskBackgroundRunner$SingleTaskBackgroundRunnerCallable.call(SingleTaskBackgroundRunner.java:449) ~[druid-indexing-service-25.0.0.jar:25.0.0] at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?] at java.lang.Thread.run(Thread.java:829) ~[?:?] Caused by: java.lang.ArrayIndexOutOfBoundsException: Index 36 out of bounds for length 2 at org.apache.avro.io.parsing.Symbol$Alternative.getSymbol(Symbol.java:460) ~[?:?] at org.apache.avro.io.ResolvingDecoder.readIndex(ResolvingDecoder.java:284) ~[?:?] at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:187) ~[?:?] at org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:136) ~[?:?] at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:247) ~[?:?] at org.apache.avro.specific.SpecificDatumReader.readRecord(SpecificDatumReader.java:123) ~[?:?] at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:179) ~[?:?] at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:160) ~[?:?] at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:187) ~[?:?] at org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:136) ~[?:?] at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:247) ~[?:?] at org.apache.avro.specific.SpecificDatumReader.readRecord(SpecificDatumReader.java:123) ~[?:?] at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:179) ~[?:?] at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:160) ~[?:?] at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:153) ~[?:?] at org.apache.druid.data.input.avro.GlobalityAvroBytesDecoder.parse(GlobalityAvroBytesDecoder.java:95) ~[?:?] ... 19 more Vamsi Gandrapu [14 hours ago](https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680756374061869?thread_ts=1680631762.258829&cid=C0309C9L90D) task does not immediately shutdown though, but get’s into StreamAppenderatorDriver and shuts down after this 2023-04-06T04:38:07,977 ERROR [task-runner-0-priority-0] org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner - Encountered exception while running task. org.apache.druid.java.util.common.parsers.ParseException: Failed to parse avro message from this origin:unknown at org.apache.druid.data.input.avro.GlobalityAvroBytesDecoder.parse(GlobalityAvroBytesDecoder.java:111) ~[?:?] at org.apache.druid.data.input.avro.AvroStreamReader.intermediateRowIterator(AvroStreamReader.java:70) ~[?:?] at org.apache.druid.data.input.IntermediateRowParsingReader.intermediateRowIteratorWithMetadata(IntermediateRowParsingReader.java:231) ~[druid-core-25.0.0.jar:25.0.0] at org.apache.druid.data.input.IntermediateRowParsingReader.read(IntermediateRowParsingReader.java:49) ~[druid-core-25.0.0.jar:25.0.0] at org.apache.druid.data.input.kafkainput.KafkaInputReader.buildBlendedRows(KafkaInputReader.java:158) ~[?:?] at org.apache.druid.data.input.kafkainput.KafkaInputReader.read(KafkaInputReader.java:129) ~[?:?] at org.apache.druid.segment.transform.TransformingInputEntityReader.read(TransformingInputEntityReader.java:43) ~[druid-processing-25.0.0.jar:25.0.0] at org.apache.druid.indexing.seekablestream.SettableByteEntityReader.read(SettableByteEntityReader.java:70) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.seekablestream.StreamChunkParser.parseWithInputFormat(StreamChunkParser.java:135) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.seekablestream.StreamChunkParser.parse(StreamChunkParser.java:104) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.runInternal(SeekableStreamIndexTaskRunner.java:635) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.run(SeekableStreamIndexTaskRunner.java:266) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTask.runTask(SeekableStreamIndexTask.java:151) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.common.task.AbstractTask.run(AbstractTask.java:169) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.overlord.SingleTaskBackgroundRunner$SingleTaskBackgroundRunnerCallable.call(SingleTaskBackgroundRunner.java:477) ~[druid-indexing-service-25.0.0.jar:25.0.0] at org.apache.druid.indexing.overlord.SingleTaskBackgroundRunner$SingleTaskBackgroundRunnerCallable.call(SingleTaskBackgroundRunner.java:449) ~[druid-indexing-service-25.0.0.jar:25.0.0] at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?] at java.lang.Thread.run(Thread.java:829) ~[?:?] Caused by: java.lang.ArrayIndexOutOfBoundsException: Index 36 out of bounds for length 2 at org.apache.avro.io.parsing.Symbol$Alternative.getSymbol(Symbol.java:460) ~[?:?] at org.apache.avro.io.ResolvingDecoder.readIndex(ResolvingDecoder.java:284) ~[?:?] at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:187) ~[?:?] at org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:136) ~[?:?] at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:247) ~[?:?] at org.apache.avro.specific.SpecificDatumReader.readRecord(SpecificDatumReader.java:123) ~[?:?] at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:179) ~[?:?] at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:160) ~[?:?] at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:187) ~[?:?] at org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:136) ~[?:?] at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:247) ~[?:?] at org.apache.avro.specific.SpecificDatumReader.readRecord(SpecificDatumReader.java:123) ~[?:?] at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:179) ~[?:?] at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:160) ~[?:?] at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:153) ~[?:?] at org.apache.druid.data.input.avro.GlobalityAvroBytesDecoder.parse(GlobalityAvroBytesDecoder.java:95) ~[?:?] ... 19 more 2023-04-06T04:38:07,984 DEBUG [task-runner-0-priority-0] org.apache.druid.indexing.common.task.AbstractTask - Not pushing task logs and reports from task. 2023-04-06T04:38:07,984 DEBUG [task-runner-0-priority-0] org.apache.druid.indexing.overlord.TaskRunnerUtils - Task [index_kafka_private.DATAPLATFORMV2.events_dcfe8936f2ad337_aepmioak] status changed to [FAILED]. 2023-04-06T04:38:07,986 INFO [task-runner-0-priority-0] org.apache.druid.indexing.worker.executor.ExecutorLifecycle - Task completed with status: { "id" : "index_kafka_private.DATAPLATFORMV2.events_dcfe8936f2ad337_aepmioak", "status" : "FAILED", "duration" : 3710, "errorMsg" : "org.apache.druid.java.util.common.parsers.ParseException: Failed to parse avro message from this ori...", "location" : { "host" : null, "port" : -1, "tlsPort" : -1 } } Vamsi Gandrapu [14 hours ago](https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1680756455411979?thread_ts=1680631762.258829&cid=C0309C9L90D) in my decoder code, I am just throwing throw new ParseException( null, false, "Failed to parse avro message from this origin:%s",origin ); ``` -- 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]
