[ksql] Handle generator raised StopIteration on query result
@@ -128,6 +128,7 @@ class KSqlApi(object):
message_data={'status': 'running', 'query_id': 1}
)
+ try:
for line in result:
# columns = line.keys()
# data.append([line[col] for col in columns])
@@ -160,6 +161,11 @@ class KSqlApi(object):
message_data={'data': data, 'meta': metadata, 'query_id': 1}
)
data = [] # TODO: special message when end of stream
+ except RuntimeError as e:
+ if 'generator raised StopIteration' in str(e):
+ return data, metadata
+ else:
+ raise e
else:
data, metadata = self._decode_result(
self.ksql(statement)
