Consumers should exit if their table is dropped. They shouldn't start…

… if the table doesn't exist.

quanta-kinesis-consumer/quanta-kinesis-consumer.go

@@ -429,6 +429,10 @@ func (m *Main) schemaChangeListener(e shared.SchemaChangeEvent) {
 	case shared.Drop:
 		//m.sessionPool.Recover(nil)
 		u.Warnf("Dropped table %s", e.Table)
+		if e.Table == m.Index {
+			u.Errorf("The table [%s], for this consumer was dropped, exiting", e.Table)
+			os.Exit(1)
+		}
 	case shared.Modify:
 		u.Warnf("Truncated table %s", e.Table)
 	case shared.Create:

shared/util.go

@@ -155,6 +155,10 @@ func putRecursive(typ reflect.Type, value reflect.Value, consul *api.Client, roo
 func UnmarshalConsul(consul *api.Client, name string) (BasicTable, error) {
 
 	table := BasicTable{Name: name}
+	keys, _, _ := consul.KV().Keys("schema/" + name, "", nil)
+	if len(keys) == 0 {
+		return table, fmt.Errorf("Table %s not found.", name)
+	}
 	ps := reflect.ValueOf(&table)
 	err := getRecursive(reflect.TypeOf(table), ps.Elem(), consul, "schema/"+name)
 	return table, err

version.txt

@@ -1 +1 @@
-quanta 0.9.10-rc-14
+quanta 0.9.10-rc-15