Skip to content

Fix race condition when initializing HealthCheckedEndpointGroup #6188

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
wants to merge 8 commits into
base: main
Choose a base branch
from
Open

Fix race condition when initializing HealthCheckedEndpointGroup #6188

Changes from 5 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
a509f46
feat : WaitUntilEndpointGroupInitialized
hyunw9 Mar 31, 2025
b92de35
feat : change EndpointGroup to Final
hyunw9 Mar 31, 2025
b7e731d
fix : correct code style
hyunw9 Apr 1, 2025
018a7b1
feat : addListener with notifyLatestValue false
hyunw9 Apr 1, 2025
31cbb14
feat : accept endpointsListener manually
hyunw9 Apr 1, 2025
d64cad0
feat : accept endpointsListener manually
hyunw9 Apr 22, 2025
8fcad43
Merge branch 'line:main' into main
hyunw9 May 12, 2025
3ba55fe
Merge branch 'line:main' into main
hyunw9 Jun 25, 2025
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
98 changes: 51 additions & 47 deletions ...java/com/linecorp/armeria/client/endpoint/healthcheck/HealthCheckedEndpointGroupTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -553,53 +553,57 @@ void cacheReflectsAttributeChanges() throws InterruptedException {
final AtomicLong updateInvokedCounter = new AtomicLong();
final Consumer<List<Endpoint>> endpointsListener = endpoints -> updateInvokedCounter.incrementAndGet();

try (HealthCheckedEndpointGroup endpointGroup =
new HealthCheckedEndpointGroup(delegate, true,
10000, 10000,
SessionProtocol.HTTP, 80,
DEFAULT_HEALTH_CHECK_RETRY_BACKOFF,
ClientOptions.of(), checkFactory,
HealthCheckStrategy.all(),
DEFAULT_ENDPOINT_PREDICATE)) {
endpointGroup.addListener(endpointsListener, true);
await().untilAsserted(() -> assertThat(updateInvokedCounter).hasValue(1));
// the counter should stay 1 after 1 second has passed
await().pollDelay(1, TimeUnit.SECONDS)
.untilAsserted(() -> assertThat(updateInvokedCounter).hasValue(1));
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.DEGRADED_ATTR))
.isFalse();
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.HEALTHY_ATTR))
.isTrue();

headers.set(ResponseHeaders.of(HttpStatus.OK, "x-envoy-degraded", ""));
// the counter should be incremented to three now
await().untilAsserted(() -> assertThat(updateInvokedCounter).hasValue(2));
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.DEGRADED_ATTR))
.isTrue();
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.HEALTHY_ATTR))
.isTrue();

// the counter should be incremented to two now
healthy.set(0);
await().untilAsserted(() -> assertThat(updateInvokedCounter).hasValue(3));
assertThat(endpointGroup.endpoints()).isEmpty();

// healthy again
healthy.set(1);
await().untilAsserted(() -> assertThat(updateInvokedCounter).hasValue(4));
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.HEALTHY_ATTR))
.isTrue();
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.DEGRADED_ATTR))
.isTrue();

// turn off degraded again
headers.set(null);
await().untilAsserted(() -> assertThat(updateInvokedCounter).hasValue(5));
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.HEALTHY_ATTR))
.isTrue();
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.DEGRADED_ATTR))
.isFalse();
}
final HealthCheckedEndpointGroup endpointGroup = new HealthCheckedEndpointGroup(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Question) Shouldn't the HealthCheckedEndpointGroup still be closed to close the ScheduledFuture at L539?

delegate, true,
10000, 10000,
SessionProtocol.HTTP, 80,
DEFAULT_HEALTH_CHECK_RETRY_BACKOFF,
ClientOptions.of(), checkFactory,
HealthCheckStrategy.all(),
DEFAULT_ENDPOINT_PREDICATE
);

endpointGroup.whenReady().join();
endpointGroup.addListener(endpointsListener, false);
endpointsListener.accept(endpointGroup.endpoints());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can't we do this?

Suggested change
endpointGroup.addListener(endpointsListener, false);
endpointsListener.accept(endpointGroup.endpoints());
endpointGroup.addListener(endpointsListener, true);

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it works ! I think Fix duplicate endpoint handling in HealthCheckedEndpointGroup
this change helps the new group include endpoints from the previous group temporarily, which avoids any brief disappearance during the transition.


await().untilAsserted(() -> assertThat(updateInvokedCounter).hasValue(1));
// the counter should stay 1 after 1 second has passed
await().pollDelay(1, TimeUnit.SECONDS)
.untilAsserted(() -> assertThat(updateInvokedCounter).hasValue(1));
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.DEGRADED_ATTR))
.isFalse();
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.HEALTHY_ATTR))
.isTrue();

headers.set(ResponseHeaders.of(HttpStatus.OK, "x-envoy-degraded", ""));
// the counter should be incremented to three now
await().untilAsserted(() -> assertThat(updateInvokedCounter).hasValue(2));
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.DEGRADED_ATTR))
.isTrue();
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.HEALTHY_ATTR))
.isTrue();

// the counter should be incremented to two now
healthy.set(0);
await().untilAsserted(() -> assertThat(updateInvokedCounter).hasValue(3));
assertThat(endpointGroup.endpoints()).isEmpty();

// healthy again
healthy.set(1);
await().untilAsserted(() -> assertThat(updateInvokedCounter).hasValue(4));
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.HEALTHY_ATTR))
.isTrue();
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.DEGRADED_ATTR))
.isTrue();

// turn off degraded again
headers.set(null);
await().untilAsserted(() -> assertThat(updateInvokedCounter).hasValue(5));
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.HEALTHY_ATTR))
.isTrue();
assertThat(endpointGroup.endpoints().get(0).attrs().attr(EndpointAttributeKeys.DEGRADED_ATTR))
.isFalse();
}

@Test
Expand Down
Loading