For use cases where we have to support multiple connectors using the same JDBC offset storage, its not a good option to delete the entire table in the save() method.

Even though the query is configurable by the user, the user is not able to pass the offset_key and has to be done manually.

 

Simple fix to modify the default DELETE query to delete records with matching offset_key

 

```
protected void save() {
try {
LOGGER.debug("Saving data to state table...");
try (PreparedStatement sqlDelete = conn.prepareStatement(config.getTableDelete())) {
sqlDelete.executeUpdate();
for (Map.Entry<String, String> mapEntry : data.entrySet()) {
Timestamp currentTs = new Timestamp(System.currentTimeMillis());
String key = (mapEntry.getKey() != null) ? mapEntry.getKey() : null;
String value = (mapEntry.getValue() != null) ? mapEntry.getValue() : null;
// Execute a query
try (PreparedStatement sql = conn.prepareStatement(config.getTableInsert()))

{ sql.setString(1, UUID.randomUUID().toString()); sql.setString(2, key); sql.setString(3, value); sql.setTimestamp(4, currentTs); sql.setInt(5, recordInsertSeq.incrementAndGet()); sql.executeUpdate(); }

}
}
conn.commit();
}```