wmedvede commented on code in PR #2214:
URL:
https://github.com/apache/incubator-kie-kogito-apps/pull/2214#discussion_r2094156500
##########
jobs-service/jobs-service-postgresql-common/src/main/java/org/kie/kogito/jobs/service/repository/postgresql/PostgreSqlJobServiceManagementRepository.java:
##########
@@ -44,69 +45,135 @@ public class PostgreSqlJobServiceManagementRepository
implements JobServiceManag
private static final Logger LOGGER =
LoggerFactory.getLogger(PostgreSqlJobServiceManagementRepository.class);
- private PgPool client;
+ private DataSource client;
@Inject
- public PostgreSqlJobServiceManagementRepository(PgPool client) {
+ public PostgreSqlJobServiceManagementRepository(DataSource client) {
this.client = client;
}
- public Uni<JobServiceManagementInfo> getAndUpdate(String id,
Function<JobServiceManagementInfo, JobServiceManagementInfo> computeUpdate) {
- LOGGER.info("get {}", id);
- return client.withTransaction(conn -> conn
- .preparedQuery("SELECT id, token, last_heartbeat FROM
job_service_management WHERE id = $1 FOR UPDATE ")
- .execute(Tuple.of(id))
- .onItem().transform(RowSet::iterator)
- .onItem().transform(iterator -> iterator.hasNext() ?
from(iterator.next()) : null)
- .onItem().invoke(r -> LOGGER.trace("got {}", r))
- .onItem().transformToUni(r -> update(conn,
computeUpdate.apply(r))));
+ private static final String GET_AND_UPDATE = "SELECT id, token,
last_heartbeat FROM job_service_management WHERE id = ? FOR UPDATE ";
+ private static final String RELEASE = "UPDATE job_service_management SET
token = null, last_heartbeat = null WHERE id = ? AND token = ? RETURNING id,
token, last_heartbeat";
+ private static final String HEARTBEAT = "UPDATE job_service_management SET
last_heartbeat = now() WHERE id = ? AND token = ? RETURNING id, token,
last_heartbeat";
+ private static final String UPDATE = "INSERT INTO job_service_management
(id, token, last_heartbeat) " +
+ "VALUES (?, ?, ?) " +
+ "ON CONFLICT (id) DO " +
+ "UPDATE SET token = ?, last_heartbeat = ? " +
+ "RETURNING id, token, last_heartbeat";
+
+ public JobServiceManagementInfo getAndUpdate(String id,
Function<JobServiceManagementInfo, JobServiceManagementInfo> computeUpdate) {
+ LOGGER.trace("get {}", id);
+ JobServiceManagementInfo jobServiceManagementInfo = null;
+ try (Connection connection = client.getConnection(); PreparedStatement
stmt = connection.prepareStatement(GET_AND_UPDATE)) {
+ stmt.setString(1, id);
+
+ ResultSet resultSet = stmt.executeQuery();
+ if (resultSet.next()) {
+ jobServiceManagementInfo = from(resultSet);
+ LOGGER.trace("got {}", jobServiceManagementInfo);
+ }
+ resultSet.close();
+ return update(connection,
computeUpdate.apply(jobServiceManagementInfo));
+ } catch (SQLException ex) {
+ LOGGER.error("Error during getAndUpdate job service management
info", ex);
+ return null;
Review Comment:
Shadows the error, callers will never know that this has failed.
At least we must throw
org.kie.kogito.jobs.service.exception.JobServiceException or we can create
another convenient RuntimeException like JobRepositoryException.
I'd recommend using
`org.kie.kogito.jobs.service.repository.JobRepositoryException` and pass the
catched exception as the root cause
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
