echonesis commented on code in PR #11789:
URL: https://github.com/apache/ozone/pull/11789#discussion_r4387128957
##########
hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/rpc/TestRpcClient.java:
##########
@@ -362,6 +369,147 @@ public void testGetOmVersionWithFutureOM() {
om(OzoneManagerVersion.UNKNOWN_VERSION))));
}
+ @Test
+ void testOmVersionRefreshesOnDemandAfterExpiry() throws Exception {
+ VersionedOmTransport transport = new VersionedOmTransport();
+ try (MockedStatic<Time> time = mockStatic(Time.class)) {
+ time.when(Time::monotonicNow).thenReturn(0L);
+ RpcClient client = createRpcClient(transport);
+ try {
+ assertEquals(OzoneManagerVersion.DEFAULT_VERSION,
client.getOmVersion());
+ transport.version = OzoneManagerVersion.SOFTWARE_VERSION;
+ time.when(Time::monotonicNow).thenReturn(TimeUnit.MINUTES.toMillis(10)
- 1);
+ assertEquals(OzoneManagerVersion.DEFAULT_VERSION,
client.getOmVersion());
+ assertEquals(1, transport.serviceInfoCalls);
+
+
time.when(Time::monotonicNow).thenReturn(TimeUnit.MINUTES.toMillis(10));
+ // Idle clients do not poll, even when their cached version has
expired.
+ assertEquals(1, transport.serviceInfoCalls);
+ assertEquals(OzoneManagerVersion.SOFTWARE_VERSION,
client.getOmVersion());
+ assertEquals(OzoneManagerVersion.SOFTWARE_VERSION,
client.getOmVersion());
+ assertEquals(2, transport.serviceInfoCalls);
+ } finally {
+ client.close();
+ }
+ }
+ }
+
+ @Test
+ void testOmVersionRefreshFailureRetainsVersionAndThrottlesRetries() throws
Exception {
+ VersionedOmTransport transport = new VersionedOmTransport();
+ try (MockedStatic<Time> time = mockStatic(Time.class)) {
+ time.when(Time::monotonicNow).thenReturn(0L);
+ RpcClient client = createRpcClient(transport);
+ try {
+ transport.failRefresh = true;
+
time.when(Time::monotonicNow).thenReturn(TimeUnit.MINUTES.toMillis(10));
+ assertEquals(OzoneManagerVersion.DEFAULT_VERSION,
client.getOmVersion());
+ assertEquals(OzoneManagerVersion.DEFAULT_VERSION,
client.getOmVersion());
+ assertEquals(2, transport.serviceInfoCalls);
+
+ transport.failRefresh = false;
+ transport.version = OzoneManagerVersion.SOFTWARE_VERSION;
+
time.when(Time::monotonicNow).thenReturn(TimeUnit.MINUTES.toMillis(20));
+ assertEquals(OzoneManagerVersion.SOFTWARE_VERSION,
client.getOmVersion());
+ assertEquals(3, transport.serviceInfoCalls);
+ } finally {
+ client.close();
+ }
+ }
+ }
+
+ @Test
+ void testFeatureCheckDiscoversFinalizedOmVersion() throws Exception {
+ VersionedOmTransport transport = new VersionedOmTransport();
+ try (MockedStatic<Time> time = mockStatic(Time.class)) {
+ time.when(Time::monotonicNow).thenReturn(0L);
+ RpcClient client = createRpcClient(transport);
+ try {
+ IOException unsupported = assertThrows(IOException.class,
+ () -> client.getObjectTagging("volume", "bucket", "key"));
+ assertThat(unsupported).hasMessageContaining("does not support S3
object tagging API");
+ transport.version = OzoneManagerVersion.SOFTWARE_VERSION;
+
time.when(Time::monotonicNow).thenReturn(TimeUnit.MINUTES.toMillis(10));
+ IOException reachedOm = assertThrows(IOException.class,
+ () -> client.getObjectTagging("volume", "bucket", "key"));
+ assertThat(reachedOm).hasMessage("Reached OM after version check");
+ assertEquals(2, transport.serviceInfoCalls);
+ } finally {
+ client.close();
+ }
+ }
+ }
+
+ @Test
+ void testConcurrentOmVersionReadersShareRefresh() throws Exception {
+ VersionedOmTransport transport = new VersionedOmTransport();
+ RpcClient client = createRpcClient(transport);
+ ExecutorService executor = Executors.newFixedThreadPool(8);
+ CountDownLatch start = new CountDownLatch(1);
+ try {
+ transport.version = OzoneManagerVersion.SOFTWARE_VERSION;
+ FieldUtils.writeField(client, "omVersionLastUpdate", Time.monotonicNow()
- TimeUnit.MINUTES.toMillis(10), true);
+ List<Future<OzoneManagerVersion>> versions = new ArrayList<>();
+ for (int i = 0; i < 8; i++) {
+ versions.add(executor.submit(() -> {
+ start.await();
+ return client.getOmVersion();
+ }));
+ }
+ start.countDown();
Review Comment:
Could we hold the first refresh inside VersionedOmTransport with a latch
while the other readers attempt getOmVersion()? The current start latch
releases the workers together, but the mocked ServiceList completes
immediately, so the test can still execute effectively sequentially.
--
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]
