Skip to content

Commit

Permalink
CAMEL-20297 camel-etcd3: do not swallow interrupted exceptions
Browse files Browse the repository at this point in the history 
This retains previous behavior, but does restore the interrupt status.
  • Loading branch information
@orpiske
orpiske committed Jan 12, 2024
1 parent a70b610 commit bc4a7ec
Showing 1 changed file with 45 additions and 10 deletions.
55 changes: 45 additions & 10 deletions ...java/org/apache/camel/component/etcd3/processor/aggregate/Etcd3AggregationRepository.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,11 @@ public Exchange add(CamelContext camelContext, String key, Exchange oldExchange,
ByteSequence.from(String.format("%s/%s", prefixName, key).getBytes()), convertToEtcd3Format(newHolder));
completablePutResponse.get();
}
} catch (InterruptedException | ExecutionException | IOException | ClassNotFoundException e) {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.error(e.getMessage(), e);
throw new OptimisticLockingException();
} catch (ExecutionException | IOException | ClassNotFoundException e) {
LOG.error(e.getMessage(), e);
throw new OptimisticLockingException();
}
Expand Down Expand Up @@ -182,7 +186,11 @@ public Exchange add(CamelContext camelContext, String key, Exchange exchange) {
PutOption.DEFAULT))
.commit()
.get();
} catch (InterruptedException | ExecutionException | IOException e) {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
} catch (ExecutionException | IOException e) {
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
}
Expand All @@ -204,7 +212,11 @@ public Set<String> scan(CamelContext camelContext) {
LOG.trace("Found {} keys for exchanges to recover in {} context", scanned.size(),
camelContext.getName());
return scanned;
} catch (InterruptedException | ExecutionException e) {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
} catch (ExecutionException e) {
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
}
Expand All @@ -226,7 +238,11 @@ public Exchange recover(CamelContext camelContext, String exchangeId) {
DefaultExchangeHolder holder
= (DefaultExchangeHolder) convertFromEtcd3Format(getResponse.getKvs().get(0).getValue());
return useRecovery ? unmarshallExchange(camelContext, holder) : null;
} catch (InterruptedException | ExecutionException | IOException | ClassNotFoundException e) {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
} catch (ExecutionException | IOException | ClassNotFoundException e) {
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
}
Expand Down Expand Up @@ -328,7 +344,11 @@ public Exchange get(CamelContext camelContext, String key) {
holder = (DefaultExchangeHolder) convertFromEtcd3Format(getResponse.getKvs().get(0).getValue());
}
return unmarshallExchange(camelContext, holder);
} catch (InterruptedException | ExecutionException | IOException | ClassNotFoundException e) {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
} catch (ExecutionException | IOException | ClassNotFoundException e) {
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
}
Expand Down Expand Up @@ -363,8 +383,11 @@ public void remove(CamelContext camelContext, String key, Exchange exchange) {
key);
throw new OptimisticLockingException();
}

} catch (InterruptedException | ExecutionException | ClassNotFoundException | IOException e) {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
} catch (ExecutionException | ClassNotFoundException | IOException e) {
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
}
Expand All @@ -382,7 +405,11 @@ public void remove(CamelContext camelContext, String key, Exchange exchange) {
LOG.trace(
"Put an exchange with ID {} for key {} into a recoverable storage in an optimistic manner.",
exchange.getExchangeId(), key);
} catch (IOException | InterruptedException | ExecutionException e) {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
} catch (IOException | ExecutionException e) {
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
}
Expand Down Expand Up @@ -425,7 +452,11 @@ public void remove(CamelContext camelContext, String key, Exchange exchange) {
.delete(ByteSequence.from(String.format("%s/%s", prefixName, key).getBytes()));
try {
completableDeleteResponse.get();
} catch (InterruptedException | ExecutionException e) {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
} catch (ExecutionException e) {
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
}
Expand All @@ -441,7 +472,11 @@ public void confirm(CamelContext camelContext, String exchangeId) {
.delete(ByteSequence.from(String.format("%s/%s", persistencePrefixName, exchangeId).getBytes()));
try {
completableDeleteResponse.get();
} catch (InterruptedException | ExecutionException e) {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
} catch (ExecutionException e) {
LOG.error(e.getMessage(), e);
throw new RuntimeCamelException(e.getMessage(), e);
}
Expand Down

0 comments on commit bc4a7ec

Please sign in to comment.