davsclaus commented on code in PR #25908:
URL: https://github.com/apache/camel/pull/25908#discussion_r3892198108
##########
components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java:
##########
@@ -129,108 +136,175 @@ public SubscriptionHelper(final SalesforceComponent
component) {
}
private MessageListener createHandshakeListener() {
- return (channel, message) ->
component.getHttpClient().getWorkerPool().execute(() -> {
- LOG.debug("[CHANNEL:META_HANDSHAKE]: {}", message);
+ return (channel, message) -> component
+ .getHttpClient()
+ .getWorkerPool()
+ .execute(
+ () -> {
+ LOG.debug("[CHANNEL:META_HANDSHAKE]: {}", message);
+
+ if (!message.isSuccessful()) {
+ LOG.warn("Handshake failure: {}", message);
+ handshakeError = (String)
message.get(ERROR_FIELD);
+ handshakeException = getFailure(message);
+ if (handshakeError != null) {
+ if (handshakeError.startsWith("403::")) {
+ String failureReason =
getFailureReason(message);
+ if
(AUTHENTICATION_INVALID.equals(failureReason)) {
+ LOG.debug(
+ "attempting login due to
handshake error: 403 -> 401::Authentication invalid");
+
session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff);
+ }
+ }
+ }
+ // failed, so keep trying with backoff
+ final long backoff =
handshakeBackoff.getAndAdd(backoffIncrement);
+ if (backoff > maxBackoff) {
+ LOG.error("Handshake retry aborted after
exceeding {} msecs backoff", maxBackoff);
+ } else {
+ LOG.debug("Pausing for {} msecs before
handshake retry", backoff);
+ if (backoff > 0) {
+ Tasks.backgroundTask()
+
.withBudget(Budgets.iterationTimeBudget()
+ .withMaxIterations(1)
+
.withInitialDelay(Duration.ofMillis(backoff))
+
.withInterval(Duration.ofMillis(1))
+
.withUnlimitedDuration()
+ .build())
+
.withScheduledExecutor(taskExecutor)
+
.withName("SalesforceHandshakeRetryDelay")
+ .build()
+
.run(component.getCamelContext(), () -> true);
+ }
+ client.handshake();
+ }
+ } else if (!channelToConsumers.isEmpty()) {
+ channelsLock.lock();
+ try {
+ channelsToSubscribe.clear();
+
channelsToSubscribe.addAll(channelToConsumers.keySet());
+ } finally {
+ channelsLock.unlock();
+ }
+ LOG.info("Handshake successful. Channels to
subscribe: {}", channelsToSubscribe);
+ }
+ });
+ }
- if (!message.isSuccessful()) {
- LOG.warn("Handshake failure: {}", message);
- handshakeError = (String) message.get(ERROR_FIELD);
- handshakeException = getFailure(message);
- if (handshakeError != null) {
- if (handshakeError.startsWith("403::")) {
- String failureReason = getFailureReason(message);
- if (AUTHENTICATION_INVALID.equals(failureReason)) {
- LOG.debug(
- "attempting login due to handshake error:
403 -> 401::Authentication invalid");
-
session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff);
- }
- }
- }
- // failed, so keep trying with backoff
- final long backoff =
handshakeBackoff.getAndAdd(backoffIncrement);
- if (backoff > maxBackoff) {
- LOG.error("Handshake retry aborted after exceeding {}
msecs backoff", maxBackoff);
- } else {
- LOG.debug("Pausing for {} msecs before handshake retry",
backoff);
- if (backoff > 0) {
- Tasks.backgroundTask()
- .withBudget(Budgets.iterationTimeBudget()
- .withMaxIterations(1)
-
.withInitialDelay(Duration.ofMillis(backoff))
- .withInterval(Duration.ofMillis(1))
- .withUnlimitedDuration()
- .build())
- .withScheduledExecutor(taskExecutor)
- .withName("SalesforceHandshakeRetryDelay")
- .build()
- .run(component.getCamelContext(), () -> true);
- }
- client.handshake();
- }
- } else if (!channelToConsumers.isEmpty()) {
- channelsLock.lock();
- try {
- channelsToSubscribe.clear();
- channelsToSubscribe.addAll(channelToConsumers.keySet());
- } finally {
- channelsLock.unlock();
+ private MessageListener createConnectionListener() {
+ return (channel, message) -> component
+ .getHttpClient()
+ .getWorkerPool()
+ .execute(
+ () -> {
+ LOG.debug("[CHANNEL:META_CONNECT]: {}", message);
+ String reconnectAdvice = message.getAdvice() !=
null
+ ? (String)
message.getAdvice().get("reconnect")
+ : null;
+
+ if (!message.isSuccessful()) {
+ LOG.warn("Connect failure: {}", message);
+ connectError = (String)
message.get(ERROR_FIELD);
+ connectException = getFailure(message);
+
+ if (connectError != null &&
connectError.equals(AUTHENTICATION_INVALID)) {
+ LOG.debug("connectError: {}",
connectError);
+ LOG.debug("Attempting login...");
+
session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff);
+ }
+ // Per Bayeux spec: handshake on null advice,
"none", "handshake", or any non-"retry" value.
+ // When advice is "retry", the CometD client
handles reconnection automatically.
+ if (reconnectAdvice == null ||
!"retry".equals(reconnectAdvice)) {
+ LOG.debug("Reconnect advice [{}] on failed
connect, initiating handshake", reconnectAdvice);
+ client.handshake();
+ } else if (isTemporaryError(message)) {
+ LOG.debug("Initiating handshake after
temporary error: {}", message);
+ client.handshake();
+ }
+ } else if (reconnectAdvice != null &&
!"retry".equals(reconnectAdvice)) {
+ LOG.warn("Reconnect advice [{}] on successful
connect, initiating handshake", reconnectAdvice);
+ client.handshake();
+ } else {
+ Set<String> toSubscribe = null;
+ channelsLock.lock();
+ try {
+ if (!channelsToSubscribe.isEmpty()) {
+ toSubscribe = new
HashSet<>(channelsToSubscribe);
+ channelsToSubscribe.clear();
+ }
+ } finally {
+ channelsLock.unlock();
+ }
+ if (toSubscribe != null) {
+ LOG.info("Subscribing to channels: {}",
toSubscribe);
+ for (var channelName : toSubscribe) {
+ var consumers =
channelToConsumers.getOrDefault(channelName, emptySet());
+ for (var consumer : consumers) {
+ subscribe(consumer);
+ }
+ }
+ }
+ }
+ });
+ }
+
+ private MessageListener createDisconnectListener() {
Review Comment:
Minor style inconsistency:
`createHandshakeListener`/`createConnectionListener` wrap their entire body in
`component.getHttpClient().getWorkerPool().execute(...)`, dispatching off the
CometD callback thread immediately. This listener instead runs its guard checks
(`isStoppingOrStopped`, the `reconnecting` CAS, the `taskExecutor` null-check)
directly on the CometD thread before handing off to `taskExecutor`. Not unsafe
(the checks are all cheap/non-blocking), just a different pattern from its two
siblings — worth a short comment on why, or aligning the style, if there's no
specific reason for the difference.
##########
components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java:
##########
@@ -129,108 +136,175 @@ public SubscriptionHelper(final SalesforceComponent
component) {
}
private MessageListener createHandshakeListener() {
- return (channel, message) ->
component.getHttpClient().getWorkerPool().execute(() -> {
- LOG.debug("[CHANNEL:META_HANDSHAKE]: {}", message);
+ return (channel, message) -> component
+ .getHttpClient()
+ .getWorkerPool()
+ .execute(
+ () -> {
+ LOG.debug("[CHANNEL:META_HANDSHAKE]: {}", message);
+
+ if (!message.isSuccessful()) {
+ LOG.warn("Handshake failure: {}", message);
+ handshakeError = (String)
message.get(ERROR_FIELD);
+ handshakeException = getFailure(message);
+ if (handshakeError != null) {
+ if (handshakeError.startsWith("403::")) {
+ String failureReason =
getFailureReason(message);
+ if
(AUTHENTICATION_INVALID.equals(failureReason)) {
+ LOG.debug(
+ "attempting login due to
handshake error: 403 -> 401::Authentication invalid");
+
session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff);
+ }
+ }
+ }
+ // failed, so keep trying with backoff
+ final long backoff =
handshakeBackoff.getAndAdd(backoffIncrement);
+ if (backoff > maxBackoff) {
+ LOG.error("Handshake retry aborted after
exceeding {} msecs backoff", maxBackoff);
+ } else {
+ LOG.debug("Pausing for {} msecs before
handshake retry", backoff);
+ if (backoff > 0) {
+ Tasks.backgroundTask()
+
.withBudget(Budgets.iterationTimeBudget()
+ .withMaxIterations(1)
+
.withInitialDelay(Duration.ofMillis(backoff))
+
.withInterval(Duration.ofMillis(1))
+
.withUnlimitedDuration()
+ .build())
+
.withScheduledExecutor(taskExecutor)
+
.withName("SalesforceHandshakeRetryDelay")
+ .build()
+
.run(component.getCamelContext(), () -> true);
+ }
+ client.handshake();
+ }
+ } else if (!channelToConsumers.isEmpty()) {
+ channelsLock.lock();
+ try {
+ channelsToSubscribe.clear();
+
channelsToSubscribe.addAll(channelToConsumers.keySet());
+ } finally {
+ channelsLock.unlock();
+ }
+ LOG.info("Handshake successful. Channels to
subscribe: {}", channelsToSubscribe);
+ }
+ });
+ }
- if (!message.isSuccessful()) {
- LOG.warn("Handshake failure: {}", message);
- handshakeError = (String) message.get(ERROR_FIELD);
- handshakeException = getFailure(message);
- if (handshakeError != null) {
- if (handshakeError.startsWith("403::")) {
- String failureReason = getFailureReason(message);
- if (AUTHENTICATION_INVALID.equals(failureReason)) {
- LOG.debug(
- "attempting login due to handshake error:
403 -> 401::Authentication invalid");
-
session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff);
- }
- }
- }
- // failed, so keep trying with backoff
- final long backoff =
handshakeBackoff.getAndAdd(backoffIncrement);
- if (backoff > maxBackoff) {
- LOG.error("Handshake retry aborted after exceeding {}
msecs backoff", maxBackoff);
- } else {
- LOG.debug("Pausing for {} msecs before handshake retry",
backoff);
- if (backoff > 0) {
- Tasks.backgroundTask()
- .withBudget(Budgets.iterationTimeBudget()
- .withMaxIterations(1)
-
.withInitialDelay(Duration.ofMillis(backoff))
- .withInterval(Duration.ofMillis(1))
- .withUnlimitedDuration()
- .build())
- .withScheduledExecutor(taskExecutor)
- .withName("SalesforceHandshakeRetryDelay")
- .build()
- .run(component.getCamelContext(), () -> true);
- }
- client.handshake();
- }
- } else if (!channelToConsumers.isEmpty()) {
- channelsLock.lock();
- try {
- channelsToSubscribe.clear();
- channelsToSubscribe.addAll(channelToConsumers.keySet());
- } finally {
- channelsLock.unlock();
+ private MessageListener createConnectionListener() {
+ return (channel, message) -> component
+ .getHttpClient()
+ .getWorkerPool()
+ .execute(
+ () -> {
+ LOG.debug("[CHANNEL:META_CONNECT]: {}", message);
+ String reconnectAdvice = message.getAdvice() !=
null
+ ? (String)
message.getAdvice().get("reconnect")
+ : null;
+
+ if (!message.isSuccessful()) {
+ LOG.warn("Connect failure: {}", message);
+ connectError = (String)
message.get(ERROR_FIELD);
+ connectException = getFailure(message);
+
+ if (connectError != null &&
connectError.equals(AUTHENTICATION_INVALID)) {
+ LOG.debug("connectError: {}",
connectError);
+ LOG.debug("Attempting login...");
+
session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff);
+ }
+ // Per Bayeux spec: handshake on null advice,
"none", "handshake", or any non-"retry" value.
+ // When advice is "retry", the CometD client
handles reconnection automatically.
+ if (reconnectAdvice == null ||
!"retry".equals(reconnectAdvice)) {
+ LOG.debug("Reconnect advice [{}] on failed
connect, initiating handshake", reconnectAdvice);
+ client.handshake();
+ } else if (isTemporaryError(message)) {
+ LOG.debug("Initiating handshake after
temporary error: {}", message);
+ client.handshake();
+ }
+ } else if (reconnectAdvice != null &&
!"retry".equals(reconnectAdvice)) {
+ LOG.warn("Reconnect advice [{}] on successful
connect, initiating handshake", reconnectAdvice);
+ client.handshake();
+ } else {
+ Set<String> toSubscribe = null;
+ channelsLock.lock();
+ try {
+ if (!channelsToSubscribe.isEmpty()) {
+ toSubscribe = new
HashSet<>(channelsToSubscribe);
+ channelsToSubscribe.clear();
+ }
+ } finally {
+ channelsLock.unlock();
+ }
+ if (toSubscribe != null) {
+ LOG.info("Subscribing to channels: {}",
toSubscribe);
+ for (var channelName : toSubscribe) {
+ var consumers =
channelToConsumers.getOrDefault(channelName, emptySet());
+ for (var consumer : consumers) {
+ subscribe(consumer);
+ }
+ }
+ }
+ }
+ });
+ }
+
+ private MessageListener createDisconnectListener() {
+ return (channel, message) -> {
+ LOG.debug("[CHANNEL:META_DISCONNECT]: {}", message);
+
+ if (isStoppingOrStopped()) {
+ LOG.debug("Ignoring disconnect message while stopping");
+ return;
+ }
+ if (!reconnecting.compareAndSet(false, true)) {
+ LOG.debug("Reconnect already in progress");
+ return;
+ }
+
+ final ScheduledExecutorService executor = taskExecutor;
+ if (executor == null) {
+ reconnecting.set(false);
+ return;
+ }
+
+ try {
+ executor.execute(this::reconnectAfterDisconnect);
+ } catch (RejectedExecutionException e) {
+ reconnecting.set(false);
+ if (!isStoppingOrStopped()) {
+ LOG.warn("Unable to schedule reconnect after server
disconnect", e);
}
- LOG.info("Handshake successful. Channels to subscribe: {}",
channelsToSubscribe);
}
- });
+ };
}
- private MessageListener createConnectionListener() {
- return (channel, message) ->
component.getHttpClient().getWorkerPool().execute(() -> {
- LOG.debug("[CHANNEL:META_CONNECT]: {}", message);
- String reconnectAdvice = message.getAdvice() != null
- ? (String) message.getAdvice().get("reconnect")
- : null;
+ private void reconnectAfterDisconnect() {
+ final BayeuxClient disconnectedClient = client;
+ try {
+ if (disconnectedClient == null || isStoppingOrStopped()) {
+ return;
+ }
- if (!message.isSuccessful()) {
- LOG.warn("Connect failure: {}", message);
- connectError = (String) message.get(ERROR_FIELD);
- connectException = getFailure(message);
-
- if (connectError != null &&
connectError.equals(AUTHENTICATION_INVALID)) {
- LOG.debug("connectError: {}", connectError);
- LOG.debug("Attempting login...");
- session.attemptLoginUntilSuccessful(backoffIncrement,
maxBackoff);
- }
- // Per Bayeux spec: handshake on null advice, "none",
"handshake", or any non-"retry" value.
- // When advice is "retry", the CometD client handles
reconnection automatically.
- if (reconnectAdvice == null ||
!"retry".equals(reconnectAdvice)) {
- LOG.debug("Reconnect advice [{}] on failed connect,
initiating handshake", reconnectAdvice);
- client.handshake();
- } else if (isTemporaryError(message)) {
- LOG.debug("Initiating handshake after temporary error:
{}", message);
- client.handshake();
- }
- } else if (reconnectAdvice != null &&
!"retry".equals(reconnectAdvice)) {
- LOG.warn("Reconnect advice [{}] on successful connect,
initiating handshake", reconnectAdvice);
- client.handshake();
- } else {
- Set<String> toSubscribe = null;
- channelsLock.lock();
- try {
- if (!channelsToSubscribe.isEmpty()) {
- toSubscribe = new HashSet<>(channelsToSubscribe);
- channelsToSubscribe.clear();
- }
- } finally {
- channelsLock.unlock();
- }
- if (toSubscribe != null) {
- LOG.info("Subscribing to channels: {}", toSubscribe);
- for (var channelName : toSubscribe) {
- var consumers =
channelToConsumers.getOrDefault(channelName, emptySet());
- for (var consumer : consumers) {
- subscribe(consumer);
- }
- }
+ final long waitMs = MILLISECONDS.convert(HANDSHAKE_TIMEOUT_SEC,
SECONDS);
+ if (!disconnectedClient.waitFor(waitMs,
BayeuxClient.State.DISCONNECTED)) {
Review Comment:
Question: `taskExecutor` is a `newSingleThreadScheduledExecutor` (see
`doStart()`). Elsewhere in this file (`createHandshakeListener`),
`taskExecutor` is only used to *schedule* a delayed callback via
`Tasks.backgroundTask()` — the actual blocking wait happens on the caller's
thread (the httpClient worker pool), so the executor's single worker thread is
only briefly occupied when the callback fires.
Here, though, `reconnectAfterDisconnect()` runs directly on `taskExecutor`'s
own worker thread (submitted via `executor.execute(...)` a few lines up), and
this `waitFor(waitMs, DISCONNECTED)` call blocks that thread for up to
`HANDSHAKE_TIMEOUT_SEC` (120s). Since the executor is single-threaded, any
handshake-retry backoff callback that becomes due on the same executor during
that window would be delayed until this wait finishes.
Given how much prior work has gone into eliminating races in this file, is
this contention window intentional/acceptable, or would it be worth moving the
blocking wait off `taskExecutor` (e.g. onto the httpClient worker pool like the
other listeners use), keeping `taskExecutor` purely for scheduling?
--
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]
