Skip to content

Fix Stale Reads on MySQL in Event Notifications DAO #142

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ThiwankaChanditha wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Fix Stale Reads on MySQL in Event Notifications DAO #142

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
96c1bac
Fix Stale Reads on MySQL in Event Notifications DAO
ThiwankaChanditha Sep 3, 2026
f8457c6
Merge branch 'wso2:main' into fix-stale-reads-mysql
ThiwankaChanditha Sep 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions .../main/java/org/wso2/dpdp/accelerator/event/notifications/dao/impl/DeliveryAckDAOImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,20 +81,27 @@ public Optional<WebhookDeliveryAck> getDeliveryAckByDeliveryId(String deliveryId
ps.setString(1, deliveryId);
try (ResultSet rs = ps.executeQuery()) {
if (rs.next()) {
return Optional.of(new WebhookDeliveryAck(
WebhookDeliveryAck ack = new WebhookDeliveryAck(
rs.getString(EventNotificationDBColumns.ACK_ID),
rs.getString(EventNotificationDBColumns.DELIVERY_ID),
rs.getTimestamp(EventNotificationDBColumns.COMPLETED_AT),
rs.getString(EventNotificationDBColumns.COMPLETION_STATUS),
rs.getString(EventNotificationDBColumns.COMPLETION_EVIDENCE)
));
);
DatabaseUtils.commitTransaction(conn);
return Optional.of(ack);
}
}
DatabaseUtils.commitTransaction(conn);
return Optional.empty();
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_GETTING_DELIVERY_ACK_BY_DELIVERY_ID, deliveryId), e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down
75 changes: 69 additions & 6 deletions ...src/main/java/org/wso2/dpdp/accelerator/event/notifications/dao/impl/DeliveryDAOImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ public Optional<WebhookDelivery> getWebhookDeliveryById(String deliveryId, Strin
ps.setString(2, orgId);
try (ResultSet rs = ps.executeQuery()) {
if (rs.next()) {
return Optional.of(new WebhookDelivery(
WebhookDelivery delivery = new WebhookDelivery(
rs.getString(EventNotificationDBColumns.DELIVERY_ID),
rs.getString(EventNotificationDBColumns.SUBSCRIPTION_ID),
rs.getString(EventNotificationDBColumns.EVENT_ID),
Expand All @@ -111,14 +111,21 @@ public Optional<WebhookDelivery> getWebhookDeliveryById(String deliveryId, Strin
rs.getTimestamp(EventNotificationDBColumns.NEXT_RETRY_AT),
rs.getTimestamp(EventNotificationDBColumns.CREATED_AT),
rs.getTimestamp(EventNotificationDBColumns.UPDATED_AT),
rs.getTimestamp(EventNotificationDBColumns.DELIVERED_AT)));
rs.getTimestamp(EventNotificationDBColumns.DELIVERED_AT));
DatabaseUtils.commitTransaction(conn);
return Optional.of(delivery);
}
}
DatabaseUtils.commitTransaction(conn);
return Optional.empty();
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_GETTING_WEBHOOK_DELIVERY, deliveryId), e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down Expand Up @@ -170,11 +177,16 @@ private List<WebhookDeliveryDispatchContext> loadDispatchContexts(String sql, in
rs.getString(EventNotificationDBColumns.TOPIC_NAME)));
}
}
DatabaseUtils.commitTransaction(conn);
return list;
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
EventNotificationCommonConstants.ERROR_GETTING_PENDING_WEBHOOK_DELIVERIES, e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down Expand Up @@ -210,11 +222,16 @@ private List<WebhookDeliveryDispatchContext> loadDispatchContextsWithCutoff(Stri
rs.getString(EventNotificationDBColumns.TOPIC_NAME)));
}
}
DatabaseUtils.commitTransaction(conn);
return list;
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
EventNotificationCommonConstants.ERROR_GETTING_PENDING_WEBHOOK_DELIVERIES, e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down Expand Up @@ -359,12 +376,17 @@ public List<WebhookDeliveryAudit> getWebhookDeliveryAudits(String deliveryId, St
rs.getTimestamp(EventNotificationDBColumns.ATTEMPT_AT)));
}
}
DatabaseUtils.commitTransaction(conn);
return list;
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_GETTING_WEBHOOK_DELIVERY_AUDITS, deliveryId),
e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down Expand Up @@ -414,22 +436,29 @@ public Optional<PollDelivery> getPollDeliveryById(String deliveryId, String orgI
ps.setString(2, orgId);
try (ResultSet rs = ps.executeQuery()) {
if (rs.next()) {
return Optional.of(new PollDelivery(
PollDelivery delivery = new PollDelivery(
rs.getString(EventNotificationDBColumns.DELIVERY_ID),
rs.getString(EventNotificationDBColumns.SUBSCRIPTION_ID),
rs.getString(EventNotificationDBColumns.EVENT_ID),
rs.getString(EventNotificationDBColumns.STATUS),
rs.getString(EventNotificationDBColumns.ERROR_CODE),
rs.getString(EventNotificationDBColumns.ERROR_DETAIL),
rs.getTimestamp(EventNotificationDBColumns.CREATED_AT),
rs.getTimestamp(EventNotificationDBColumns.COMPLETED_AT)));
rs.getTimestamp(EventNotificationDBColumns.COMPLETED_AT));
DatabaseUtils.commitTransaction(conn);
return Optional.of(delivery);
}
}
DatabaseUtils.commitTransaction(conn);
return Optional.empty();
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_GETTING_POLL_DELIVERY, deliveryId), e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down Expand Up @@ -465,12 +494,17 @@ public List<PollDelivery> getPendingPollDeliveries(String orgId, String groupId,
rs.getTimestamp(EventNotificationDBColumns.COMPLETED_AT)));
}
}
DatabaseUtils.commitTransaction(conn);
return candidates;
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_GETTING_PENDING_POLL_DELIVERIES,
subscriptionId), e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down Expand Up @@ -766,13 +800,18 @@ public List<SubscriptionDeliverySummary> listSubscriptionDeliveries(String orgId
}
}
}
DatabaseUtils.commitTransaction(conn);
return list;
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_LISTING_DELIVERIES_FOR_SUBSCRIPTION,
subscriptionId),
e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand All @@ -791,14 +830,21 @@ public Optional<SubscriptionDeliverySummary> getSubscriptionDeliveryById(String
ps.setString(6, orgId);
try (ResultSet rs = ps.executeQuery()) {
if (rs.next()) {
return Optional.of(mapSummary(rs));
SubscriptionDeliverySummary summary = mapSummary(rs);
DatabaseUtils.commitTransaction(conn);
return Optional.of(summary);
}
}
DatabaseUtils.commitTransaction(conn);
return Optional.empty();
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_GETTING_SUBSCRIPTION_DELIVERY, deliveryId), e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down Expand Up @@ -906,11 +952,16 @@ public List<SubscriptionDeliverySummary> listOrgDeliveries(String orgId, String
}
}
}
DatabaseUtils.commitTransaction(conn);
return list;
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_LISTING_ORG_DELIVERIES, orgId), e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand All @@ -926,14 +977,21 @@ public Optional<SubscriptionDeliverySummary> getOrgDeliveryById(String orgId, St
ps.setString(3, deliveryId);
try (ResultSet rs = ps.executeQuery()) {
if (rs.next()) {
return Optional.of(mapSummary(rs));
SubscriptionDeliverySummary summary = mapSummary(rs);
DatabaseUtils.commitTransaction(conn);
return Optional.of(summary);
}
}
DatabaseUtils.commitTransaction(conn);
return Optional.empty();
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_GETTING_ORG_DELIVERY, deliveryId), e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down Expand Up @@ -993,11 +1051,16 @@ public List<SubscriptionDeliverySummary> listEventDeliveries(String orgId, Strin
}
}
}
DatabaseUtils.commitTransaction(conn);
return list;
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_LISTING_ORG_DELIVERIES, orgId), e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down
30 changes: 27 additions & 3 deletions ...ao/src/main/java/org/wso2/dpdp/accelerator/event/notifications/dao/impl/EventDAOImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,14 +91,20 @@ public Optional<Event> getEventById(String eventId, String orgId) {
if (rs.next()) {
Event event = mapEvent(rs);
event.setPurposes(getEventPurposes(conn, eventId));
DatabaseUtils.commitTransaction(conn);
return Optional.of(event);
}
}
DatabaseUtils.commitTransaction(conn);
return Optional.empty();
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_GETTING_EVENT_BY_ID, eventId), e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down Expand Up @@ -135,7 +141,12 @@ public void addEventPurposes(Connection conn, String eventId, List<String> purpo
public List<String> getEventPurposes(String eventId) {
Connection conn = DatabaseUtils.getDBConnection();
try {
return getEventPurposes(conn, eventId);
List<String> purposes = getEventPurposes(conn, eventId);
DatabaseUtils.commitTransaction(conn);
return purposes;
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down Expand Up @@ -164,13 +175,20 @@ public boolean hasActiveEventsForTopic(String topicId) {
try {
try (PreparedStatement ps = conn.prepareStatement(getQueries(conn).getHasActiveEventsForTopicQuery())) {
ps.setString(1, topicId);
boolean hasActive;
try (ResultSet rs = ps.executeQuery()) {
return rs.next();
hasActive = rs.next();
}
DatabaseUtils.commitTransaction(conn);
return hasActive;
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_HAS_ACTIVE_EVENTS_FOR_TOPIC, topicId), e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down Expand Up @@ -232,11 +250,17 @@ public PaginatedDAOResult<Event> searchEvents(String orgId, String topic, String
}
}

return new PaginatedDAOResult<>(events, total[0]);
PaginatedDAOResult<Event> result = new PaginatedDAOResult<>(events, total[0]);
DatabaseUtils.commitTransaction(conn);
return result;
} catch (SQLException e) {
DatabaseUtils.rollbackTransaction(conn);
throw new EventNotificationDataAccessException(
String.format(EventNotificationCommonConstants.ERROR_LISTING_EVENTS, orgId), e);
}
} catch (RuntimeException e) {
DatabaseUtils.rollbackTransaction(conn);
throw e;
} finally {
DatabaseUtils.closeConnection(conn);
}
Expand Down
Loading