AmatyaAvadhanula commented on code in PR #14271:
URL: https://github.com/apache/druid/pull/14271#discussion_r1214290956
##########
server/src/main/java/org/apache/druid/metadata/SQLMetadataStorageActionHandler.java:
##########
@@ -165,34 +166,61 @@ public void insert(
{
try {
getConnector().retryWithHandle(
- (HandleCallback<Void>) handle -> {
- final String sql = StringUtils.format(
- "INSERT INTO %s (id, created_date, datasource, payload, type,
group_id, active, status_payload) "
- + "VALUES (:id, :created_date, :datasource, :payload, :type,
:group_id, :active, :status_payload)",
- getEntryTable()
- );
- handle.createStatement(sql)
- .bind("id", id)
- .bind("created_date", timestamp.toString())
- .bind("datasource", dataSource)
- .bind("payload", jsonMapper.writeValueAsBytes(entry))
- .bind("type", type)
- .bind("group_id", groupId)
- .bind("active", active)
- .bind("status_payload", jsonMapper.writeValueAsBytes(status))
- .execute();
- return null;
- },
- e -> getConnector().isTransientException(e) &&
!(isStatementException(e) && getEntry(id).isPresent())
+ handle -> insertEntryWithHandle(handle, id, timestamp, dataSource,
entry, active, status, type, groupId),
+ this::isTransientDruidException
);
}
+ catch (CallbackFailedException e) {
+ propageAsRuntimeException(e.getCause());
+ }
catch (Exception e) {
- if (isStatementException(e) && getEntry(id).isPresent()) {
- throw new EntryExistsException(id, e);
- } else {
- Throwables.propagateIfPossible(e);
- throw new RuntimeException(e);
- }
+ propageAsRuntimeException(e);
+ }
+ }
+
+ private void propageAsRuntimeException(Throwable t)
+ {
+ Throwables.propagateIfPossible(t);
+ throw new RuntimeException(t);
+ }
+
+ /**
+ * Inserts the given entry into the metadata store. Any exception thrown by a
+ * HandleCallback is wrapped in a {@link DruidException} wrapped in a
+ * {@link CallbackFailedException}.
+ */
+ private int insertEntryWithHandle(
Review Comment:
Nit: It doesn't seem like the return value is being used. Can this be Void
instead?
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java:
##########
@@ -222,25 +222,19 @@ public Response taskPost(final Task task, @Context final
HttpServletRequest req)
return asLeaderWith(
taskMaster.getTaskQueue(),
- new Function<TaskQueue, Response>()
- {
- @Override
- public Response apply(TaskQueue taskQueue)
- {
- try {
- taskQueue.add(task);
- return Response.ok(ImmutableMap.of("task",
task.getId())).build();
- }
- catch (EntryExistsException e) {
- return Response.status(Response.Status.BAD_REQUEST)
- .entity(
- ImmutableMap.of(
- "error",
- StringUtils.format("Task[%s] already
exists!", task.getId())
- )
- )
- .build();
- }
+ taskQueue -> {
+ try {
+ taskQueue.add(task);
+ return Response.ok(ImmutableMap.of("task", task.getId())).build();
+ }
+ catch (DruidException e) {
+ return Response.status(e.getResponseCode())
+ .entity(ImmutableMap.of("error", e.getMessage()))
+ .build();
+ }
+ catch (Exception e) {
+ log.info("Got final exception [%s]", e.getClass());
Review Comment:
Logging the taskId as well might help
##########
server/src/main/java/org/apache/druid/metadata/SQLMetadataStorageActionHandler.java:
##########
@@ -165,34 +166,61 @@ public void insert(
{
try {
getConnector().retryWithHandle(
- (HandleCallback<Void>) handle -> {
- final String sql = StringUtils.format(
- "INSERT INTO %s (id, created_date, datasource, payload, type,
group_id, active, status_payload) "
- + "VALUES (:id, :created_date, :datasource, :payload, :type,
:group_id, :active, :status_payload)",
- getEntryTable()
- );
- handle.createStatement(sql)
- .bind("id", id)
- .bind("created_date", timestamp.toString())
- .bind("datasource", dataSource)
- .bind("payload", jsonMapper.writeValueAsBytes(entry))
- .bind("type", type)
- .bind("group_id", groupId)
- .bind("active", active)
- .bind("status_payload", jsonMapper.writeValueAsBytes(status))
- .execute();
- return null;
- },
- e -> getConnector().isTransientException(e) &&
!(isStatementException(e) && getEntry(id).isPresent())
+ handle -> insertEntryWithHandle(handle, id, timestamp, dataSource,
entry, active, status, type, groupId),
+ this::isTransientDruidException
);
}
+ catch (CallbackFailedException e) {
+ propageAsRuntimeException(e.getCause());
+ }
catch (Exception e) {
- if (isStatementException(e) && getEntry(id).isPresent()) {
- throw new EntryExistsException(id, e);
- } else {
- Throwables.propagateIfPossible(e);
- throw new RuntimeException(e);
- }
+ propageAsRuntimeException(e);
+ }
+ }
+
+ private void propageAsRuntimeException(Throwable t)
Review Comment:
typo? propagateAsRuntimeException
--
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]
