Impala should not crash for invalid avro serialized data

Here is the backtrace when impala crashed with an invalid data type.

(gdb) bt
#0 0x000000334f6aca3d in nanosleep () from /home/anuj/cdh38698/tmp/imp_dir/libc.so.6
#1 0x000000334f6ac8b0 in sleep () from /home/anuj/cdh38698/tmp/imp_dir/libc.so.6
#2 0x00007f0b1cb83bd2 in os::infinite_sleep() () from /home/anuj/cdh38698/tmp/imp_dir/libjvm.so
#3 0x00007f0b1cd089b4 in VMError::report_and_die() () from /home/anuj/cdh38698/tmp/imp_dir/libjvm.so
#4 0x00007f0b1cd0950e in crash_handler(int, siginfo*, void*) () from /home/anuj/cdh38698/tmp/imp_dir/libjvm.so
#5 0x00007f0b1cb87bf2 in os::Linux::chained_handler(int, siginfo*, void*) () from /home/anuj/cdh38698/tmp/imp_dir/libjvm.so
#6 0x00007f0b1cb8d8d6 in JVM_handle_linux_signal () from /home/anuj/cdh38698/tmp/imp_dir/libjvm.so
#7 <signal handler called>
#8 impala::ReadWriteUtil::ReadZLong (buf=0x7f0979792390) at /usr/src/debug/impala-2.3.0-cdh5.5.2/be/src/exec/read-write-util.cc:30
#9 0x0000000000c49e52 in ReadZInt (buf=<optimized out>) at /usr/src/debug/impala-2.3.0-cdh5.5.2/be/src/exec/read-write-util.h:217
#10 impala::HdfsAvroScanner::ReadAvroInt32 (this=<optimized out>, type=impala::INVALID_TYPE, data=<optimized out>, write_slot=false, slot=0x0, pool=<optimized out>)
at /usr/src/debug/impala-2.3.0-cdh5.5.2/be/src/exec/hdfs-avro-scanner-ir.cc:61
#11 0x0000000000c4224a in impala::HdfsAvroScanner::MaterializeTuple (this=0x7f0a30af1b80, record_schema=..., pool=0x7bda8fa0, data=0x7f0979792390, tuple=0x15373d98)
at /usr/src/debug/impala-2.3.0-cdh5.5.2/be/src/exec/hdfs-avro-scanner.cc:542
#12 0x0000000000c4a09e in impala::HdfsAvroScanner::DecodeAvroData (this=0x7f0a30af1b80, max_tuples=43, pool=0x7bda8fa0, data=0x7f0979792390, tuple=0x15373d98, tuple_row=0xb7932b8)
at /usr/src/debug/impala-2.3.0-cdh5.5.2/be/src/exec/hdfs-avro-scanner-ir.cc:28
#13 0x0000000000c4481d in impala::HdfsAvroScanner::ProcessRange (this=0x7f0a30af1b80) at /usr/src/debug/impala-2.3.0-cdh5.5.2/be/src/exec/hdfs-avro-scanner.cc:496
#14 0x0000000000cc8a22 in impala::BaseSequenceScanner::ProcessSplit (this=0x7f0a30af1b80) at /usr/src/debug/impala-2.3.0-cdh5.5.2/be/src/exec/base-sequence-scanner.cc:144
#15 0x0000000000c27dfc in impala::HdfsScanNode::ScannerThread (this=0xa4f5000) at /usr/src/debug/impala-2.3.0-cdh5.5.2/be/src/exec/hdfs-scan-node.cc:946
#16 0x0000000000ae0faa in operator() (this=<optimized out>) at /opt/toolchain/boost-pic-1.55.0/include/boost/function/function_template.hpp:767
#17 impala::Thread::SuperviseThread(std::string const&, std::string const&, boost::function<void ()>, impala::Promise<long>*) (name=..., category=..., functor=..., thread_started=0x7f0982da0420)
at /usr/src/debug/impala-2.3.0-cdh5.5.2/be/src/util/thread.cc:314
#18 0x0000000000ae3250 in operator()<void (const std::string&, const std::string&, impala::Thread::ThreadFunctor, impala::Promise<long int>*), boost::_bi::list0> (a=...,
f=@0x76cd6fb8: 0xae0df0 <impala::Thread::SuperviseThread(std::string const&, std::string const&, boost::function<void ()>, impala::Promise<long>*)>, this=0x76cd6fc0)
at /opt/toolchain/boost-pic-1.55.0/include/boost/bind/bind.hpp:457
#19 operator() (this=0x76cd6fb8) at /opt/toolchain/boost-pic-1.55.0/include/boost/bind/bind_template.hpp:20
#20 boost::detail::thread_data<boost::_bi::bind_t<void, void (std::string const&, std::string const&, boost::function<void ()>, impala::Promise<long>), boost::_bi::list4<boost::_bi::value<std::string>, boost::_bi::value<std::string>, boost::_bi::value<boost::function<void ()> >, boost::_bi::value<impala::Promise<long>> > > >::run() (this=0x76cd6e00)
at /opt/toolchain/boost-pic-1.55.0/include/boost/thread/detail/thread.hpp:117
#21 0x0000000000d28c43 in ?? ()
#22 0x00000033502079d1 in start_thread () from /home/anuj/cdh38698/tmp/imp_dir/libpthread.so.0
#23 0x000000334f6e88fd in clone () from /home/anuj/cdh38698/tmp/imp_dir/libc.so.6

void HdfsAvroScanner::ReadAvroInt32(PrimitiveType type, uint8_t** data, bool write_slot,
void* slot, MemPool* pool) {
int32_t val = ReadWriteUtil::ReadZInt(data);
if (write_slot) {
if (type == TYPE_INT) {
reinterpret_cast<int32_t>(slot) = val;
} else if (type == TYPE_BIGINT) {
reinterpret_cast<int64_t>(slot) = val;
} else if (type == TYPE_FLOAT) {
reinterpret_cast<float>(slot) = val;
} else if (type == TYPE_DOUBLE) {
reinterpret_cast<double>(slot) = val;
} else {
DCHECK(false); <<<<<
}
}
}

I am not sure what the expected behavior should be but it should not crash.