Kafka indexing service failed when the message is compressed with snappy in 0.10.1-rc4

In 0.10.1-rc4, kafka indexing service use the kafka-client-0.10.2. If the producer compress the message with snappy, here comes the error:

2017-08-21T17:13:57,499 ERROR [task-runner-0-priority-0] io.druid.indexing.overlord.ThreadPoolTaskRunner - Exception while running task[KafkaIndexTask{id=index_kafka_nsky_biz_monitor_order_20ba66af32953e8_akggpjah, type=index_kafka, dataSource=nsky_biz_monitor_order}]
org.apache.kafka.common.KafkaException: java.lang.ClassNotFoundException: org.xerial.snappy.SnappyInputStream
        at org.apache.kafka.common.record.MemoryRecordsBuilder.wrapForInput(MemoryRecordsBuilder.java:441) ~[kafka-clients-0.10.2.0.jar:?]
        at org.apache.kafka.common.record.RecordsIterator$DeepRecordsIterator.<init>(RecordsIterator.java:150) ~[kafka-clients-0.10.2.0.jar:?]
        at org.apache.kafka.common.record.RecordsIterator.makeNext(RecordsIterator.java:79) ~[kafka-clients-0.10.2.0.jar:?]
        at org.apache.kafka.common.record.RecordsIterator.makeNext(RecordsIterator.java:34) ~[kafka-clients-0.10.2.0.jar:?]
        at org.apache.kafka.common.utils.AbstractIterator.maybeComputeNext(AbstractIterator.java:79) ~[kafka-clients-0.10.2.0.jar:?]
        at org.apache.kafka.common.utils.AbstractIterator.hasNext(AbstractIterator.java:45) ~[kafka-clients-0.10.2.0.jar:?]
        at org.apache.kafka.clients.consumer.internals.Fetcher.parseCompletedFetch(Fetcher.java:772) ~[kafka-clients-0.10.2.0.jar:?]
        at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:473) ~[kafka-clients-0.10.2.0.jar:?]
        at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1062) ~[kafka-clients-0.10.2.0.jar:?]
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:995) ~[kafka-clients-0.10.2.0.jar:?]
        at io.druid.indexing.kafka.KafkaIndexTask.run(KafkaIndexTask.java:397) ~[?:?]
        at io.druid.indexing.overlord.ThreadPoolTaskRunner$ThreadPoolTaskRunnerCallable.call(ThreadPoolTaskRunner.java:436) [druid-indexing-service-0.10.1-rc4-SNAPSHOT.jar:0.10.1-rc4-SNAPSHOT]
        at io.druid.indexing.overlord.ThreadPoolTaskRunner$ThreadPoolTaskRunnerCallable.call(ThreadPoolTaskRunner.java:408) [druid-indexing-service-0.10.1-rc4-SNAPSHOT.jar:0.10.1-rc4-SNAPSHOT]
        at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_65]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_65]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_65]
        at java.lang.Thread.run(Thread.java:745) [?:1.8.0_65]
Caused by: java.lang.ClassNotFoundException: org.xerial.snappy.SnappyInputStream
        at java.net.URLClassLoader.findClass(URLClassLoader.java:381) ~[?:1.8.0_65]
        at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[?:1.8.0_65]
        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331) ~[?:1.8.0_65]
        at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[?:1.8.0_65]
        at java.lang.Class.forName0(Native Method) ~[?:1.8.0_65]
        at java.lang.Class.forName(Class.java:264) ~[?:1.8.0_65]
        at org.apache.kafka.common.record.MemoryRecordsBuilder$3.get(MemoryRecordsBuilder.java:75) ~[kafka-clients-0.10.2.0.jar:?]
        at org.apache.kafka.common.record.MemoryRecordsBuilder$MemoizingConstructorSupplier.get(MemoryRecordsBuilder.java:477) ~[kafka-clients-0.10.2.0.jar:?]
        at org.apache.kafka.common.record.MemoryRecordsBuilder.wrapForInput(MemoryRecordsBuilder.java:438) ~[kafka-clients-0.10.2.0.jar:?]
        ... 16 more

``

``

MemoryRecordsBuilder wants load org.xerial.snappy.SnappyOutputStream with AppClassLoader is can't find the class.
If put the snappy-java-1.1.2.6.jar in lib, it will work. But I'm not sure it is a good way.

Hmm, I guess we should include snappy in the kafka-indexing-service extension. A patch would be welcome.