Skip to content

Commit 9016c56

Browse files
m-nagarajanm-nagarajan
committed
addressed review comments
1 parent 8b42f7f commit 9016c56

3 files changed

Lines changed: 220 additions & 29 deletions

File tree

‎clients/venice-client/src/test/java/com/linkedin/venice/fastclient/meta/InstanceHealthMonitorTest.java‎

Lines changed: 87 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -160,7 +160,13 @@ public void testUnhealthyNodeDetectedByHB() throws Exception {
160160
CompletableFuture<TransportClientResponse> requestFuture = new CompletableFuture<>();
161161
ChainedCompletableFuture<Integer, Integer> chainedRequestFuture =
162162
monitor.trackHealthBasedOnRequestToInstance(instance, requestFuture);
163-
Thread.sleep(1500); // must exceed routingRequestDefaultTimeoutMS (1000ms) to trigger unhealthy detection
163+
// Wait for the routing request to time out (it completes the request future) before completing the request,
164+
// so the timeout is not cancelled and the instance is marked suspicious.
165+
TestUtils.waitForNonDeterministicAssertion(
166+
5,
167+
TimeUnit.SECONDS,
168+
true,
169+
() -> assertTrue(requestFuture.isCompletedExceptionally(), "routing request should have timed out"));
164170
requestFuture.complete(null);
165171
chainedRequestFuture.getOriginalFuture().complete(SC_GONE);
166172
// Pending request counter will be reset with a delay
@@ -198,38 +204,14 @@ public void testUnhealthyNodeDetectedByHB() throws Exception {
198204
*/
199205
@Test
200206
public void testUpdateLiveInstanceSetEvictsHostRemovedFromFleet() throws Exception {
201-
Map<String, Long> requestPathToResponseDelayMap = new VeniceConcurrentHashMap<>();
202-
Map<String, CompletableFuture<RestResponse>> requestPathToResponseFutureMap = new VeniceConcurrentHashMap<>();
203-
CompletableFuture<RestResponse> hbResponseFuture =
204-
CompletableFuture.completedFuture(new RestResponseBuilder().setStatus(SC_OK).build());
205-
String hbPath = instance + "/" + QueryAction.HEALTH.toString().toLowerCase();
206-
requestPathToResponseFutureMap.put(hbPath, hbResponseFuture);
207-
// a large delay forces every heartbeat to this instance to time out, so it keeps failing
208-
requestPathToResponseDelayMap.put(hbPath, 10000L);
209-
MockClient client = new MockClient(requestPathToResponseDelayMap, requestPathToResponseFutureMap);
210-
211-
InstanceHealthMonitorConfig config = InstanceHealthMonitorConfig.builder()
212-
.setRoutingRequestDefaultTimeoutMS(1000L)
213-
.setRoutingPendingRequestCounterInstanceBlockThreshold(10)
214-
.setHeartBeatIntervalSeconds(1)
215-
.setHeartBeatRequestTimeoutMS(100L)
216-
.setRoutingTimedOutRequestCounterResetDelayMS(2000)
217-
.setClient(client)
218-
.build();
219-
220-
try (InstanceHealthMonitor monitor = new InstanceHealthMonitor(config)) {
221-
// Drive the instance into the unhealthy set via a timed-out user request + a failing heartbeat.
222-
CompletableFuture<TransportClientResponse> requestFuture = new CompletableFuture<>();
223-
ChainedCompletableFuture<Integer, Integer> chainedRequestFuture =
224-
monitor.trackHealthBasedOnRequestToInstance(instance, requestFuture);
225-
Thread.sleep(1500); // must exceed routingRequestDefaultTimeoutMS (1000ms) to trigger unhealthy detection
226-
requestFuture.complete(null);
227-
chainedRequestFuture.getOriginalFuture().complete(SC_GONE);
207+
try (InstanceHealthMonitor monitor = new InstanceHealthMonitor(failingHeartbeatConfig())) {
208+
// A request to the host hangs, so the routing timeout + failing heartbeat mark it unhealthy.
209+
monitor.trackHealthBasedOnRequestToInstance(instance, new CompletableFuture<>());
228210
TestUtils.waitForNonDeterministicAssertion(
229211
5,
230212
TimeUnit.SECONDS,
231213
true,
232-
() -> assertFalse(monitor.isInstanceHealthy(instance), "instance should be marked unhealthy"));
214+
() -> assertFalse(monitor.isInstanceHealthy(instance), "slow host should be marked unhealthy"));
233215
assertEquals(monitor.getUnhealthyInstanceCount(), 1);
234216

235217
// A refresh whose serving set still contains the host keeps it tracked (a live-but-unhealthy host).
@@ -255,6 +237,64 @@ public void testUpdateLiveInstanceSetEvictsHostRemovedFromFleet() throws Excepti
255237
}
256238
}
257239

240+
/**
241+
* Unhealthy-host lifecycle: a slow host is marked unhealthy by the monitor, evicted when it leaves the serving set,
242+
* then rejoins clean.
243+
*/
244+
@Test
245+
public void testUnhealthyHostEvictedThenRejoinsClean() throws Exception {
246+
try (InstanceHealthMonitor monitor = new InstanceHealthMonitor(failingHeartbeatConfig())) {
247+
// Host is in the cluster; a request to it hangs, so the routing timeout + failing heartbeat mark it unhealthy.
248+
monitor.updateLiveInstanceSet(Collections.singleton(instance));
249+
monitor.trackHealthBasedOnRequestToInstance(instance, new CompletableFuture<>());
250+
TestUtils.waitForNonDeterministicAssertion(5, TimeUnit.SECONDS, true, () -> {
251+
assertFalse(monitor.isInstanceHealthy(instance), "slow host should be marked unhealthy");
252+
assertEquals(monitor.getUnhealthyInstanceCount(), 1);
253+
});
254+
255+
// Host leaves the cluster: the next serving set drops it, so the monitor evicts it.
256+
monitor.updateLiveInstanceSet(Collections.singleton("https://other.host:4321"));
257+
TestUtils.waitForNonDeterministicAssertion(5, TimeUnit.SECONDS, true, () -> {
258+
assertTrue(monitor.isInstanceHealthy(instance), "departed host should no longer be tracked");
259+
assertEquals(monitor.getUnhealthyInstanceCount(), 0);
260+
});
261+
262+
// The same host returns and starts clean.
263+
monitor.updateLiveInstanceSet(Collections.singleton(instance));
264+
assertTrue(monitor.isInstanceHealthy(instance), "rejoined host should start clean");
265+
assertEquals(monitor.getUnhealthyInstanceCount(), 0);
266+
}
267+
}
268+
269+
/**
270+
* Healthy-host lifecycle: even a host that never went unhealthy has its tracking state evicted when it leaves the
271+
* serving set, and rejoins clean.
272+
*/
273+
@Test
274+
public void testHealthyHostEvictedThenRejoinsClean() throws Exception {
275+
InstanceHealthMonitorConfig config =
276+
InstanceHealthMonitorConfig.builder().setRoutingRequestDefaultTimeoutMS(10000L).build();
277+
try (InstanceHealthMonitor monitor = new InstanceHealthMonitor(config)) {
278+
// Host is in the cluster and serves a successful request, so it is tracked (a drained counter) and healthy.
279+
monitor.updateLiveInstanceSet(Collections.singleton(instance));
280+
ChainedCompletableFuture<Integer, Integer> chainedFuture = monitor.trackHealthBasedOnRequestToInstance(instance);
281+
chainedFuture.getOriginalFuture().complete(SC_OK);
282+
waitQuietly(chainedFuture.getResultFuture());
283+
assertTrue(monitor.isInstanceHealthy(instance));
284+
assertTrue(monitor.hasPendingRequestCounter(instance));
285+
286+
// Host leaves the cluster: even though it is healthy, its tracking state is evicted.
287+
monitor.updateLiveInstanceSet(Collections.singleton("https://other.host:4321"));
288+
assertTrue(monitor.isInstanceHealthy(instance));
289+
assertFalse(monitor.hasPendingRequestCounter(instance));
290+
291+
// The same host returns clean.
292+
monitor.updateLiveInstanceSet(Collections.singleton(instance));
293+
assertTrue(monitor.isInstanceHealthy(instance));
294+
assertFalse(monitor.hasPendingRequestCounter(instance));
295+
}
296+
}
297+
258298
/**
259299
* A drained (zero) pending-request counter for a departed host is evicted; a counter for a live host or one with an
260300
* in-flight request is kept so accounting stays correct.
@@ -292,6 +332,24 @@ public void testUpdateLiveInstanceSetEvictsDrainedPendingRequestCounters() throw
292332
}
293333
}
294334

335+
/** Config whose heartbeat to {@link #instance} always times out, so the instance stays unhealthy once tracked. */
336+
private static InstanceHealthMonitorConfig failingHeartbeatConfig() {
337+
Map<String, CompletableFuture<RestResponse>> futureMap = new VeniceConcurrentHashMap<>();
338+
Map<String, Long> delayMap = new VeniceConcurrentHashMap<>();
339+
String hbPath = instance + "/" + QueryAction.HEALTH.toString().toLowerCase();
340+
futureMap.put(hbPath, CompletableFuture.completedFuture(new RestResponseBuilder().setStatus(SC_OK).build()));
341+
// a large delay forces every heartbeat to this instance to time out, so it keeps failing
342+
delayMap.put(hbPath, 10000L);
343+
return InstanceHealthMonitorConfig.builder()
344+
.setRoutingRequestDefaultTimeoutMS(1000L)
345+
.setRoutingPendingRequestCounterInstanceBlockThreshold(10)
346+
.setHeartBeatIntervalSeconds(1)
347+
.setHeartBeatRequestTimeoutMS(100L)
348+
.setRoutingTimedOutRequestCounterResetDelayMS(2000)
349+
.setClient(new MockClient(delayMap, futureMap))
350+
.build();
351+
}
352+
295353
private void waitQuietly(CompletableFuture future) throws InterruptedException {
296354
try {
297355
future.get();

‎clients/venice-client/src/test/java/com/linkedin/venice/fastclient/meta/RequestBasedMetadataTest.java‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import static org.testng.Assert.assertSame;
2929
import static org.testng.Assert.assertTrue;
3030

31+
import com.linkedin.r2.transport.common.Client;
3132
import com.linkedin.venice.client.exceptions.VeniceClientException;
3233
import com.linkedin.venice.client.exceptions.VeniceClientHttpException;
3334
import com.linkedin.venice.client.schema.RouterBackedSchemaReader;
@@ -157,6 +158,69 @@ public void testUpdateCacheReconcilesHealthMonitorWithServingSet() throws IOExce
157158
}
158159
}
159160

161+
/**
162+
* Host lifecycle exercised through the real {@link RequestBasedMetadata} refresh + reconciliation and a real
163+
* {@link InstanceHealthMonitor} (transport and metadata responses are mocked): a hung request marks a host
164+
* unhealthy, a refresh that drops the host evicts it, and a refresh that re-adds it leaves it clean.
165+
*/
166+
@Test(timeOut = TEST_TIMEOUT)
167+
public void testHostLifecycleThroughMetadataRefreshAndRequestPath() throws IOException, InterruptedException {
168+
String storeName = "testStore";
169+
// No-op refresh executor so only this test drives refreshes (via updateCache) and the metadata sequence is
170+
// consumed in order.
171+
ScheduledExecutorService noOpRefreshExecutor = mock(ScheduledExecutorService.class);
172+
ClientConfig clientConfig =
173+
RequestBasedMetadataTestUtils.getMockClientConfig(storeName, false, false, noOpRefreshExecutor);
174+
175+
// Real monitor whose heartbeat client never responds, so a suspicious instance becomes unhealthy on the next beat.
176+
InstanceHealthMonitor realMonitor = new InstanceHealthMonitor(
177+
InstanceHealthMonitorConfig.builder()
178+
.setRoutingRequestDefaultTimeoutMS(1000L)
179+
.setHeartBeatIntervalSeconds(1)
180+
.setHeartBeatRequestTimeoutMS(100L)
181+
.setRoutingTimedOutRequestCounterResetDelayMS(2000)
182+
.setClient(mock(Client.class))
183+
.build());
184+
doReturn(realMonitor).when(clientConfig).getInstanceHealthMonitor();
185+
186+
RequestBasedMetadata requestBasedMetadata = null;
187+
try {
188+
D2TransportClient d2TransportClient = RequestBasedMetadataTestUtils.getMockD2TransportClientCycling(storeName);
189+
requestBasedMetadata = new RequestBasedMetadata(clientConfig, d2TransportClient);
190+
requestBasedMetadata
191+
.setMetadataResponseSchemaReader(RequestBasedMetadataTestUtils.getMockRouterBackedSchemaReader());
192+
requestBasedMetadata.setD2ServiceDiscovery(getMockD2ServiceDiscovery(d2TransportClient, storeName));
193+
194+
// State 1: REPLICA1_NAME and REPLICA2_NAME serving (the first refresh runs synchronously in start()).
195+
requestBasedMetadata.start();
196+
197+
// A hung request to REPLICA1_NAME: the routing timeout marks it suspicious and the failing heartbeat then marks
198+
// it unhealthy.
199+
requestBasedMetadata
200+
.trackHealthBasedOnRequestToInstance(REPLICA1_NAME, CURRENT_VERSION, 0, new CompletableFuture<>());
201+
waitForNonDeterministicAssertion(15, TimeUnit.SECONDS, true, () -> {
202+
assertFalse(realMonitor.isInstanceHealthy(REPLICA1_NAME));
203+
assertEquals(realMonitor.getUnhealthyInstanceCount(), 1);
204+
});
205+
206+
// State 2: a refresh drops REPLICA1_NAME, so reconciliation evicts it and it is healthy again.
207+
requestBasedMetadata.updateCache(false);
208+
waitForNonDeterministicAssertion(15, TimeUnit.SECONDS, true, () -> {
209+
assertTrue(realMonitor.isInstanceHealthy(REPLICA1_NAME));
210+
assertEquals(realMonitor.getUnhealthyInstanceCount(), 0);
211+
});
212+
213+
// State 3: a refresh re-adds REPLICA1_NAME; it returns clean.
214+
requestBasedMetadata.updateCache(false);
215+
assertTrue(realMonitor.isInstanceHealthy(REPLICA1_NAME));
216+
assertEquals(realMonitor.getUnhealthyInstanceCount(), 0);
217+
} finally {
218+
if (requestBasedMetadata != null) {
219+
requestBasedMetadata.close();
220+
}
221+
}
222+
}
223+
160224
/**
161225
* This is to test warmUpInstancesFutures in case if the one of the conn warmup fails
162226
*

‎clients/venice-client/src/test/java/com/linkedin/venice/fastclient/meta/RequestBasedMetadataTestUtils.java‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -201,6 +201,75 @@ public static D2TransportClient getMockD2TransportClient(
201201
return d2TransportClient;
202202
}
203203

204+
/**
205+
* A {@link D2TransportClient} whose METADATA endpoint returns three responses in sequence, driving REPLICA1_NAME
206+
* through present -> absent (replaced by NEW_REPLICA_NAME) -> present across three refreshes.
207+
*/
208+
public static D2TransportClient getMockD2TransportClientCycling(String storeName) {
209+
D2TransportClient d2TransportClient = mock(D2TransportClient.class);
210+
211+
Map<CharSequence, List<CharSequence>> withReplica1 = new HashMap<>();
212+
withReplica1.put("0", Collections.singletonList(REPLICA1_NAME));
213+
withReplica1.put("1", Collections.singletonList(REPLICA2_NAME));
214+
Map<CharSequence, List<CharSequence>> withoutReplica1 = new HashMap<>();
215+
withoutReplica1.put("0", Collections.singletonList(NEW_REPLICA_NAME));
216+
withoutReplica1.put("1", Collections.singletonList(REPLICA2_NAME));
217+
218+
int metadataResponseSchemaId = AvroProtocolDefinition.SERVER_METADATA_RESPONSE.getCurrentProtocolVersion();
219+
CompletableFuture<TransportClientResponse> state1 = CompletableFuture.completedFuture(
220+
new TransportClientResponse(
221+
metadataResponseSchemaId,
222+
CompressionStrategy.NO_OP,
223+
buildCyclingMetadataBody(withReplica1)));
224+
CompletableFuture<TransportClientResponse> state2 = CompletableFuture.completedFuture(
225+
new TransportClientResponse(
226+
metadataResponseSchemaId + 1,
227+
CompressionStrategy.NO_OP,
228+
buildCyclingMetadataBody(withoutReplica1)));
229+
CompletableFuture<TransportClientResponse> state3 = CompletableFuture.completedFuture(
230+
new TransportClientResponse(
231+
metadataResponseSchemaId,
232+
CompressionStrategy.NO_OP,
233+
buildCyclingMetadataBody(withReplica1)));
234+
235+
when(d2TransportClient.get(eq(QueryAction.METADATA.toString().toLowerCase() + "/" + storeName)))
236+
.thenReturn(state1, state2, state3);
237+
238+
TransportClientResponse dictionaryResponse = new TransportClientResponse(0, CompressionStrategy.NO_OP, DICTIONARY);
239+
doReturn(CompletableFuture.completedFuture(dictionaryResponse)).when(d2TransportClient)
240+
.get(eq(QueryAction.DICTIONARY.toString().toLowerCase() + "/" + storeName + "/" + CURRENT_VERSION));
241+
242+
return d2TransportClient;
243+
}
244+
245+
private static byte[] buildCyclingMetadataBody(Map<CharSequence, List<CharSequence>> routeMap) {
246+
Map<String, String> partitionerParams = new HashMap<>();
247+
partitionerParams.put("testKey", "testValue");
248+
VersionProperties versionProperties = new VersionProperties(
249+
CURRENT_VERSION,
250+
CompressionStrategy.ZSTD_WITH_DICT.getValue(),
251+
2,
252+
"com.linkedin.venice.partitioner.DefaultVenicePartitioner",
253+
Collections.unmodifiableMap(partitionerParams),
254+
1);
255+
Map<CharSequence, Integer> helixGroupMap = new HashMap<>();
256+
helixGroupMap.put(REPLICA1_NAME, 0);
257+
helixGroupMap.put(REPLICA2_NAME, 1);
258+
helixGroupMap.put(NEW_REPLICA_NAME, 0);
259+
MetadataResponseRecord metadataResponse = new MetadataResponseRecord(
260+
versionProperties,
261+
Collections.singletonList(CURRENT_VERSION),
262+
Collections.singletonMap("1", KEY_SCHEMA),
263+
Collections.singletonMap("1", VALUE_SCHEMA),
264+
1,
265+
routeMap,
266+
helixGroupMap,
267+
150,
268+
ExternalStorageReadMode.VENICE_ONLY.getValue());
269+
return SerializerDeserializerFactory.getAvroGenericSerializer(MetadataResponseRecord.SCHEMA$)
270+
.serialize(metadataResponse);
271+
}
272+
204273
public static D2ServiceDiscovery getMockD2ServiceDiscovery(D2TransportClient d2TransportClient, String storeName) {
205274
D2ServiceDiscovery d2ServiceDiscovery = mock(D2ServiceDiscovery.class);
206275
D2ServiceDiscoveryResponse d2ServiceDiscoveryResponse = new D2ServiceDiscoveryResponse();

0 commit comments

Comments
 (0)