Kafka Consumer generate locks database transaction
@@ -522,20 +522,37 @@ public class Update extends BaseTransform<UpdateMeta, UpdateData>
return false;
}
+ @Override
+ public void batchComplete() throws HopException {
+ // This is needed to properly handle transactions when we are using SingleThreadedExecutor
+ commitBatch(false);
+ }
+
@Override
public void dispose() {
+ // This is needed to properly handle transactions when we are using SingleThreadedExecutor
+ commitBatch(true);
+ super.dispose();
+ }
+ private void commitBatch(boolean dispose) {
if (data.db != null) {
try {
if (!data.db.isAutoCommit()) {
if (getErrors() == 0) {
+ if (dispose) {
data.db.emptyAndCommit(data.prepStatementUpdate, meta.useBatchUpdate());
+ } else {
+ data.db.commit();
+ }
} else {
data.db.rollback();
}
}
+ if (dispose) {
data.db.closePreparedStatement(data.prepStatementUpdate);
data.db.closePreparedStatement(data.prepStatementLookup);
+ }
} catch (HopDatabaseException e) {
logError(
BaseMessages.getString(PKG, "Update.Log.UnableToCommitUpdateConnection")
@@ -544,9 +561,10 @@ public class Update extends BaseTransform<UpdateMeta, UpdateData>
+ e.toString());
setErrors(1);
} finally {
+ if (dispose)
data.db.disconnect();
}
}
- super.dispose();
+
}
}
