diff --git a/be/src/exec/kudu-table-sink.cc b/be/src/exec/kudu-table-sink.cc
index 541d398..87eeb84 100644
--- a/be/src/exec/kudu-table-sink.cc
+++ b/be/src/exec/kudu-table-sink.cc
@@ -137,6 +137,25 @@
   KUDU_RETURN_IF_ERROR(client_->OpenTable(table_desc_->table_name(), &table_),
       "Unable to open Kudu table");
 
+  // Verify the KuduTable's schema is what we expect, in case it was modified since
+  // analysis. If the underlying schema is changed after this point but before the write
+  // completes, the KuduTable's schema will stay the same and we'll get an error back from
+  // the Kudu server.
+  for (int i = 0; i < output_expr_evals_.size(); ++i) {
+    int col_idx = kudu_table_sink_.referenced_columns.empty() ?
+        i : kudu_table_sink_.referenced_columns[i];
+    if (col_idx >= table_->schema().num_columns()) {
+      return Status(strings::Substitute(
+          "Table $0 has fewer columns than expected.", table_desc_->name()));
+    }
+    ColumnType type = KuduDataTypeToColumnType(table_->schema().Column(col_idx).type());
+    if (type != output_expr_evals_[i]->root().type()) {
+      return Status(strings::Substitute("Column $0 has unexpected type. ($1 vs. $2)",
+          table_->schema().Column(col_idx).name(), type.DebugString(),
+          output_expr_evals_[i]->root().type().DebugString()));
+    }
+  }
+
   session_ = client_->NewSession();
   session_->SetTimeoutMillis(FLAGS_kudu_operation_timeout_ms);
 
diff --git a/be/src/exec/kudu-util.cc b/be/src/exec/kudu-util.cc
index 6688b7e..e5a822a 100644
--- a/be/src/exec/kudu-util.cc
+++ b/be/src/exec/kudu-util.cc
@@ -34,6 +34,7 @@
 using kudu::client::KuduClient;
 using kudu::client::KuduClientBuilder;
 using kudu::client::KuduColumnSchema;
+using DataType = kudu::client::KuduColumnSchema::DataType;
 
 DECLARE_bool(disable_kudu);
 DECLARE_int32(kudu_operation_timeout_ms);
@@ -177,4 +178,21 @@
 
   return Status::OK();
 }
+
+ColumnType KuduDataTypeToColumnType(DataType type) {
+  switch (type) {
+    case DataType::INT8: return ColumnType(PrimitiveType::TYPE_TINYINT);
+    case DataType::INT16: return ColumnType(PrimitiveType::TYPE_SMALLINT);
+    case DataType::INT32: return ColumnType(PrimitiveType::TYPE_INT);
+    case DataType::INT64: return ColumnType(PrimitiveType::TYPE_BIGINT);
+    case DataType::STRING: return ColumnType(PrimitiveType::TYPE_STRING);
+    case DataType::BOOL: return ColumnType(PrimitiveType::TYPE_BOOLEAN);
+    case DataType::FLOAT: return ColumnType(PrimitiveType::TYPE_FLOAT);
+    case DataType::DOUBLE: return ColumnType(PrimitiveType::TYPE_DOUBLE);
+    case DataType::BINARY: return ColumnType(PrimitiveType::TYPE_BINARY);
+    case DataType::UNIXTIME_MICROS: return ColumnType(PrimitiveType::TYPE_TIMESTAMP);
+  }
+  return ColumnType(PrimitiveType::INVALID_TYPE);
+}
+
 }  // namespace impala
diff --git a/be/src/exec/kudu-util.h b/be/src/exec/kudu-util.h
index 4d9b77b..28c6b27 100644
--- a/be/src/exec/kudu-util.h
+++ b/be/src/exec/kudu-util.h
@@ -78,5 +78,8 @@
 Status WriteKuduValue(int col, PrimitiveType type, const void* value,
     bool copy_strings, kudu::KuduPartialRow* row) WARN_UNUSED_RESULT;
 
+/// Takes a Kudu client DataType and returns the corresponding Impala ColumnType.
+ColumnType KuduDataTypeToColumnType(kudu::client::KuduColumnSchema::DataType type);
+
 } /// namespace impala
 #endif
diff --git a/tests/query_test/test_kudu.py b/tests/query_test/test_kudu.py
index 9715b4e..b951237 100644
--- a/tests/query_test/test_kudu.py
+++ b/tests/query_test/test_kudu.py
@@ -30,7 +30,10 @@
 from kudu.client import Partitioning
 import logging
 import pytest
+import random
 import textwrap
+import threading
+import time
 from datetime import datetime
 from pytz import utc
 
@@ -395,6 +398,48 @@
     cursor.execute("select count(*) from %s" % table_name)
     print cursor.fetchall() == [(i, )]
 
+  def test_concurrent_schema_change(self, cursor, unique_database):
+    """Tests that an insert into a Kudu table with a concurrent schema change either
+    succeeds or fails gracefully."""
+    table_name = "%s.test_schema_change" % unique_database
+    cursor.execute("""create table %s (col0 bigint primary key, col1 bigint)
+    partition by hash(col0) partitions 16 stored as kudu""" % table_name)
+
+    iters = 5
+    def insert_values():
+      threading.current_thread().errors = []
+      client = self.create_impala_client()
+      for i in range(0, iters):
+        time.sleep(random.random()) # sleeps for up to one second
+        try:
+          client.execute("insert into %s values (0, 0), (1, 1)" % table_name)
+        except Exception as e:
+          threading.current_thread().errors.append(e)
+
+    insert_thread = threading.Thread(target=insert_values)
+    insert_thread.start()
+
+    for i in range(0, iters):
+      time.sleep(random.random()) # sleeps for up to one second
+      cursor.execute("alter table %s drop column col1" % table_name)
+      if i % 2 == 0:
+        cursor.execute("alter table %s add columns (col1 string)" % table_name)
+      else:
+        cursor.execute("alter table %s add columns (col1 bigint)" % table_name)
+
+    insert_thread.join()
+
+    for error in insert_thread.errors:
+      msg = str(error)
+      # The first two are AnalysisExceptions, the next two come from KuduTableSink::Open()
+      # if the schema has changed since analysis, the last comes from the Kudu server if
+      # the schema changes between KuduTableSink::Open() and when the write ops are sent.
+      assert "has fewer columns (1) than the SELECT / VALUES clause returns (2)" in msg \
+        or "(type: TINYINT) is not compatible with column 'col1' (type: STRING)" in msg \
+        or "has fewer columns than expected." in msg \
+        or "Column col1 has unexpected type." in msg \
+        or "Client provided column col1[int64 NULLABLE] not present in tablet" in msg
+
 class TestCreateExternalTable(KuduTestSuite):
 
   def test_external_timestamp_default_value(self, cursor, kudu_client, unique_database):