Repository: hazelcast/hazelcast (language: java)

Issue:
com.hazelcast.jet.core.ObservableResultsTest.configureCapacity[1: mode=client] [HZ-3420]
PR builder failed on `master`:
https://jenkins.hazelcast.com/job/Hazelcast-pr-builder/18957/

<details><summary>Stacktrace:</summary>

```
java.lang.AssertionError: expected:<class java.lang.IllegalStateException> but was:<class java.lang.RuntimeException>
	at org.junit.Assert.fail(Assert.java:89)
	at org.junit.Assert.failNotEquals(Assert.java:835)
	at org.junit.Assert.assertEquals(Assert.java:120)
	at org.junit.Assert.assertEquals(Assert.java:146)
	at com.hazelcast.jet.core.ObservableResultsTest.assertThrowsException(ObservableResultsTest.java:611)
	at com.hazelcast.jet.core.ObservableResultsTest.configureCapacity(ObservableResultsTest.java:526)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at com.hazelcast.test.FailOnTimeoutStatement$CallableStatement.call(FailOnTimeoutStatement.java:115)
	at com.hazelcast.test.FailOnTimeoutStatement$CallableStatement.call(FailOnTimeoutStatement.java:107)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.lang.Thread.run(Thread.java:834)
```

</details>

<details><summary>Standard output:</summary>

```
14:48:08,949  INFO || - [MasterJobContext] hz.ObservableResultsTest_jolly_hertz.cached.thread-14 - [127.0.0.1]:5701 [dev] [5.4.0-SNAPSHOT] Execution of job '0a02-d70f-d68b-0001', execution 0a02-d705-0921-0001 completed successfully
	Start time: 2023-06-14T14:48:08.942
	Duration: 00:00:00.007
	To see additional job metrics enable JobConfig.storeMetricsAfterJobCompletion
14:48:08,950 ERROR || - [AddDynamicConfigOperation] hz.ObservableResultsTest_jolly_hertz.generic-operation.thread-0 - [127.0.0.1]:5701 [dev] [5.4.0-SNAPSHOT] Cannot add a dynamic configuration 'RingbufferConfig{name='__jet.observables.2891bc91-ec3d-49e4-819b-d2de2ac00fb4', capacity=30000, backupCount=1, asyncBackupCount=0, timeToLiveSeconds=0, inMemoryFormat=BINARY, ringbufferStoreConfig=RingbufferStoreConfig{enabled=false, className='null', storeImplementation=null, factoryClassName='null', factoryImplementation=null, properties={}}, splitBrainProtectionName=null, mergePolicyConfig=MergePolicyConfig{policy='com.hazelcast.spi.merge.PutIfAbsentMergePolicy', batchSize=100}}' as there is already a conflicting configuration 'RingbufferConfig{name='__jet.observables.2891bc91-ec3d-49e4-819b-d2de2ac00fb4', capacity=20000, backupCount=1, asyncBackupCount=0, timeToLiveSeconds=0, inMemoryFormat=BINARY, ringbufferStoreConfig=RingbufferStoreConfig{enabled=false, className='null', storeImplementation=null, factoryClassName='null', factoryImplementation=null, properties={}}, splitBrainProtectionName=null, mergePolicyConfig=MergePolicyConfig{policy='com.hazelcast.spi.merge.PutIfAbsentMergePolicy', batchSize=100}}'
com.hazelcast.config.InvalidConfigurationException: Cannot add a dynamic configuration 'RingbufferConfig{name='__jet.observables.2891bc91-ec3d-49e4-819b-d2de2ac00fb4', capacity=30000, backupCount=1, asyncBackupCount=0, timeToLiveSeconds=0, inMemoryFormat=BINARY, ringbufferStoreConfig=RingbufferStoreConfig{enabled=false, className='null', storeImplementation=null, factoryClassName='null', factoryImplementation=null, properties={}}, splitBrainProtectionName=null, mergePolicyConfig=MergePolicyConfig{policy='com.hazelcast.spi.merge.PutIfAbsentMergePolicy', batchSize=100}}' as there is already a conflicting configuration 'RingbufferConfig{name='__jet.observables.2891bc91-ec3d-49e4-819b-d2de2ac00fb4', capacity=20000, backupCount=1, asyncBackupCount=0, timeToLiveSeconds=0, inMemoryFormat=BINARY, ringbufferStoreConfig=RingbufferStoreConfig{enabled=false, className='null', storeImplementation=null, factoryClassName='null', factoryImplementation=null, properties={}}, splitBrainProtectionName=null, mergePolicyConfig=MergePolicyConfig{policy='com.hazelcast.spi.merge.PutIfAbsentMergePolicy', batchSize=100}}'
	at com.hazelcast.internal.dynamicconfig.ClusterWideConfigurationService.checkCurrentConfigNullOrEqual(ClusterWideConfigurationService.java:356) ~[classes/:?]
	at com.hazelcast.internal.dynamicconfig.ClusterWideConfigurationService.registerConfigLocally(ClusterWideConfigurationService.java:340) ~[classes/:?]
	at com.hazelcast.internal.dynamicconfig.AddDynamicConfigOperation.run(AddDynamicConfigOperation.java:48) ~[classes/:?]
	at com.hazelcast.spi.impl.operationservice.Operation.call(Operation.java:193) [classes/:?]
	at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.call(OperationRunnerImpl.java:299) [classes/:?]
	at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:270) [classes/:?]
	at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:502) [classes/:?]
	at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:202) [classes/:?]
	at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:142) [classes/:?]
	at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.loop(OperationThread.java:134) [classes/:?]
	at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.executeRun(OperationThread.java:115) [classes/:?]
	at com.hazelcast.internal.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:111) [classes/:?]
14:48:08,952 DEBUG || - [JobCoordinationService] hz.ObservableResultsTest_jolly_hertz.cached.thread-14 - [127.0.0.1]:5701 [dev] [5.4.0-SNAPSHOT] job '0a02-d70f-d68b-0001', execution 0a02-d705-0921-0001 is completed
14:48:11,001 DEBUG || - [JobRepository] hz.ObservableResultsTest_jolly_hertz.cached.thread-9 - [127.0.0.1]:5701 [dev] [5.4.0-SNAPSHOT] Skipping job cleanup because job records IMap does not exist
14:48:11,051  INFO |configureCapacity[1: mode=client]| - [HazelcastTestSupport] Time-limited test - Shutting down node factory as @After action
BuildInfo right after configureCapacity[1: mode=client](com.hazelcast.jet.core.ObservableResultsTest): BuildInfo{version='5.4.0-SNAPSHOT', build='20230614', buildNumber=20230614, revision=33f25fd, enterprise=false, serializationVersion=1}
Hiccups measured while running test 'configureCapacity[1: mode=client](com.hazelcast.jet.core.ObservableResultsTest):'
14:48:05, accumulated pauses: 2076 ms, max pause: 2037 ms, pauses over 1000 ms: 1
14:48:10, accumulated pauses: 0 ms, max pause: 0 ms, pauses over 1000 ms: 0
```

</details>



Discussion / hints:
It seems like "ringbufferExists" method invoked in the "configureCapacity" method of the ObservableImpl produced the wrong result due to getDistributedObject() method missing the just-created ring buffer.
Internal Jira issue: [HZ-3420](https://hazelcast.atlassian.net//browse/HZ-3420)

Fix patch:
diff --git a/hazelcast/src/main/java/com/hazelcast/jet/impl/observer/ObservableImpl.java b/hazelcast/src/main/java/com/hazelcast/jet/impl/observer/ObservableImpl.java
index f439cdbc3e568..70a4c850597f5 100644
--- a/hazelcast/src/main/java/com/hazelcast/jet/impl/observer/ObservableImpl.java
+++ b/hazelcast/src/main/java/com/hazelcast/jet/impl/observer/ObservableImpl.java
@@ -19,6 +19,7 @@
 import com.hazelcast.client.HazelcastClientNotActiveException;
 import com.hazelcast.client.impl.clientside.HazelcastClientInstanceImpl;
 import com.hazelcast.config.Config;
+import com.hazelcast.config.InvalidConfigurationException;
 import com.hazelcast.config.RingbufferConfig;
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.core.HazelcastInstanceNotActiveException;
@@ -105,16 +106,26 @@ public void removeObserver(@Nonnull UUID registrationId) {
     @Override
     public Observable<T> configureCapacity(int capacity) {
         String ringbufferName = ringbufferName(name);
-        if (ringbufferExists(ringbufferName)) {
-            throw new IllegalStateException("Underlying buffer for observable '" + name + "' is already created.");
-        }
-        Config config = hzInstance.getConfig();
-        try {
-            config.addRingBufferConfig(new RingbufferConfig(ringbufferName).setCapacity(capacity));
-        } catch (Exception e) {
-            throw new RuntimeException("Failed configuring capacity: " + e, e);
+        int i = 0;
+        while (true) {
+            i++;
+            if (ringbufferExists(ringbufferName)) {
+                throw new IllegalStateException("Underlying buffer for observable '" + name + "' is already created.");
+            }
+            Config config = hzInstance.getConfig();
+            try {
+                config.addRingBufferConfig(new RingbufferConfig(ringbufferName).setCapacity(capacity));
+                return this;
+            } catch (InvalidConfigurationException e) {
+                logger.warning("Configuration for ringbuffer " + name + " already exists. "
+                        + "Maybe the ringbuffer was missed due to unreliable HazelcastInstance#getDistributedObjects");
+                if (i == 2) {
+                    throw new RuntimeException("Failed configuring capacity: " + e, e);
+                }
+            } catch (Exception e) {
+                throw new RuntimeException("Failed configuring capacity: " + e, e);
+            }
         }
-        return this;
     }
 
     @Override


Test patch:
diff --git a/hazelcast/src/test/java/com/hazelcast/jet/core/ObservableResultsTest.java b/hazelcast/src/test/java/com/hazelcast/jet/core/ObservableResultsTest.java
index e911dc0eb4889..320b5b8236140 100644
--- a/hazelcast/src/test/java/com/hazelcast/jet/core/ObservableResultsTest.java
+++ b/hazelcast/src/test/java/com/hazelcast/jet/core/ObservableResultsTest.java
@@ -54,12 +54,12 @@
 import java.util.stream.LongStream;
 import java.util.stream.Stream;
 
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
 
 @Category({QuickTest.class, ParallelJVMTest.class})
 public class ObservableResultsTest extends TestInClusterSupport {
@@ -517,19 +517,19 @@ public void configureCapacity() {
                 .writeTo(Sinks.observable(o));
         //then
         o.configureCapacity(20_000); //still possible, pipeline not executing yet
-        assertThrowsException(o::getConfiguredCapacity, IllegalStateException.class);
+        assertThatThrownBy(o::getConfiguredCapacity).isInstanceOf(IllegalStateException.class);
 
         //when
         Job job = hz().getJet().newJob(pipeline);
         assertExecutionStarted(job);
         //then
-        assertThrowsException(() -> o.configureCapacity(30_000), IllegalStateException.class);
+        assertThatThrownBy(() -> o.configureCapacity(30_000)).isInstanceOf(IllegalStateException.class);
         assertEquals(20_000, o.getConfiguredCapacity());
 
         //when
         job.join();
         ///then
-        assertThrowsException(() -> o.configureCapacity(30_000), IllegalStateException.class);
+        assertThatThrownBy(() -> o.configureCapacity(30_000)).isInstanceOf(IllegalStateException.class);
         assertEquals(20_000, o.getConfiguredCapacity());
     }
 
@@ -537,7 +537,7 @@ public void configureCapacity() {
     public void configureCapacityMultipleTimes() {
         Observable<Object> o = hz().getJet().newObservable();
         o.configureCapacity(10);
-        assertThrowsException(() -> o.configureCapacity(20), RuntimeException.class);
+        assertThatThrownBy(() -> o.configureCapacity(20)).isInstanceOf(RuntimeException.class);
     }
 
     @Test
@@ -602,16 +602,6 @@ private void assertExecutionStarted(Job job) {
                 || JobStatus.COMPLETED.equals(job.getStatus())));
     }
 
-    private void assertThrowsException(Runnable action, Class<? extends Throwable> exceptionClass) {
-        //then
-        try {
-            action.run();
-            fail("Expected exception not thrown");
-        } catch (Throwable t) {
-            assertEquals(exceptionClass, t.getClass());
-        }
-    }
-
     private static void assertSortedValues(TestObserver observer, Long... values) {
         assertTrueEventually(() -> assertEquals(Arrays.asList(values), observer.getSortedValues()));
     }