Fix method is not safely accessed by multiple concurrent threads.

[shibd]

Motivation

#13663 Flaky-test: org.apache.pulsar.metadata.LockManagerTest.updateValue

The root cause it that MetadataCacheImpl#refresh method is not safely accessed by multiple concurrent threads.

pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java

Lines 262 to 269 in 2a7515f

	@Override
	public void refresh(String path) {
	// Refresh object of path if only it is cached before.
	if (objCache.getIfPresent(path) != null) {
	objCache.synchronous().invalidate(path);
	objCache.synchronous().refresh(path);
	}
	}

The AbstractMetadaStore.put method will have two threads refresh the cache in parallel.

1. (thread 1)Callback method of storePut returned:
   

   pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java

   Lines 279 to 281 in 2a7515f

   	metadataCaches.forEach(c -> c.refresh(path));
   	return stat;

2. (thread 2)Notification implement in storePut internal(ZK, RocksDB, LocalMemory):

pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java

Lines 183 to 193 in 2a7515f

	return CompletableFuture.supplyAsync(() -> {
	listeners.forEach(listener -> {
	try {
	listener.accept(notification);
	} catch (Throwable t) {
	log.error("Failed to process metadata store notification", t);
	}
	});
	return null;
	}, executor);

We can only get the execution result of thread 1 on the client, and can't wait for thread 2.

When thread 2 has not finished refreshing the last time, At this point, if the update starts again, the old value may be returned. Reference by caffeine note

LoadingCache#refresh

  /**
   * Loads a new value for the {@code key}, asynchronously. While the new value is loading the
   * previous value (if any) will continue to be returned by {@code get(key)} unless it is evicted.
   * If the new value is loaded successfully it will replace the previous value in the cache; if an
   * exception is thrown while refreshing the previous value will remain, <i>and the exception will
   * be logged (using {@link java.util.logging.Logger}) and swallowed</i>.
   * <p>
   * Caches loaded by a {@link CacheLoader} will call {@link CacheLoader#reload} if the cache
   * currently contains a value for the {@code key}, and {@link CacheLoader#load} otherwise. Loading
   * is asynchronous by delegating to the default executor.
   *
   * @param key key with which a value may be associated
   * @throws NullPointerException if the specified key is null
   */
  void refresh(@NonNull K key);

Cache#invalidate

  /**
   * Discards any cached value for the {@code key}. The behavior of this operation is undefined for
   * an entry that is being loaded (or reloaded) and is otherwise not present.
   *
   * @param key the key whose mapping is to be removed from the cache
   * @throws NullPointerException if the specified key is null
   */
  void invalidate(@NonNull @CompatibleWith("K") Object key);

In this unit test, the first execution of the following method will trigger the update cache, and thread 2 may not complete the update all the time.

pulsar/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LockManagerTest.java

Lines 142 to 144 in 2a7515f

	ResourceLock<String> lock = lockManager.acquireLock("/my/path/1", "lock-1").join();
	assertEquals(lock.getValue(), "lock-1");
	assertEquals(cache.get("/my/path/1").join().get(), "lock-1");

When the value is updated again, it is possible that the cache update did not succeed.So you may get the last cached value.

May be getValue is equals "locak-1"

pulsar/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LockManagerTest.java

Lines 146 to 148 in 2a7515f

	lock.updateValue("value-2").join();
	assertEquals(lock.getValue(), "value-2");
	assertEquals(cache.get("/my/path/1").join().get(), "value-2");

I solved it directly with synchronous lock. After many tests, the problem no longer appears. If there is a better implementation, it can be discussed. Thank you~

Modifications

• add synchronized to MetadataCacheImpl#refresh、MetadataCacheImpl#invalidate、MetadataCacheImpl#invalidateAll

Documentation

• no-need-doc

[shibd]

@Jason918 @Technoboy- @merlimat Can you help me review it? Thanks.

[shibd]

/pulsarbot run-failure-checks

[Jason918]

Sorry, I am not getting this yet.

In the error part.

pulsar/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LockManagerTest.java

Lines 146 to 148 in 2a7515f

	lock.updateValue("value-2").join();
	assertEquals(lock.getValue(), "value-2");
	assertEquals(cache.get("/my/path/1").join().get(), "value-2");

We have the following key code line executed in the order:

[0] [Main Thread] lock.updateValue("value-2").join(); 

    [1. Main Thread] store.put(path, payload, ...)

    [2. metadata-store Thread] metadataCaches.forEach(c -> c.refresh(path));

        [3. metadata-store Thread] objCache.synchronous().invalidate(path);

        [4. metadata-store Thread] objCache.synchronous().refresh(path);

[5] cache.get("/my/path/1").join()

Is this the right executing order that [3] happens before [5]?
Why would [5] got value before [0] ?

[shibd]

Sorry, I am not getting this yet.

In the error part.

pulsar/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LockManagerTest.java

Lines 146 to 148 in 2a7515f

	lock.updateValue("value-2").join();
	assertEquals(lock.getValue(), "value-2");
	assertEquals(cache.get("/my/path/1").join().get(), "value-2");

We have the following key code line executed in the order:

[0] [Main Thread] lock.updateValue("value-2").join(); 

    [1. Main Thread] store.put(path, payload, ...)

    [2. metadata-store Thread] metadataCaches.forEach(c -> c.refresh(path));

        [3. metadata-store Thread] objCache.synchronous().invalidate(path);

        [4. metadata-store Thread] objCache.synchronous().refresh(path);

[5] cache.get("/my/path/1").join()

Is this the right executing order that [3] happens before [5]? Why would [5] got value before [0] ?

@Jason918 Your description is only the execution order of thread 1. If only thread 1 is working, it is no problem.

However, there is another thread in refresh during actual operation(thread 2), which is triggered by the notification of metadata. This is performed in the background. We don't know when to complete it

pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java

Lines 181 to 197 in 2a7515f

	protected CompletableFuture<Void> receivedNotification(Notification notification) {
	try {
	return CompletableFuture.supplyAsync(() -> {
	listeners.forEach(listener -> {
	try {
	listener.accept(notification);
	} catch (Throwable t) {
	log.error("Failed to process metadata store notification", t);
	}
	});
	return null;
	}, executor);
	} catch (RejectedExecutionException e) {
	return FutureUtil.failedFuture(e);
	}
	}

I use thread 1 and thread 2 to express two update flow. You can look at the above explanation.

[Jason918]

I use thread 1 and thread 2 to express two update flow. You can look at the above explanation.

I saw the explanation above and sorry that I missed the key info about "Cache#invalidate", it can't invalid loading keys. This is the root cause. I think the async loading is ok if the invalidate can work with loading keys.

[Jason918]

About the solution, I still have some concerns.

In my understanding, the org.apache.pulsar.metadata.cache.impl.MetadataCacheImpl#refresh should be always called in the "metadata-store" Thread. Something else must be wrong with callback handling. It does not go through org.apache.pulsar.metadata.impl.AbstractMetadataStore#execute.
We should fix that instead of adding synchronized for refresh.

[Jason918]

Checked codes, we can fix this by wrap zkc.multi callback into execute in org.apache.pulsar.metadata.impl.ZKMetadataStore#batchOperation.

You can check org.apache.pulsar.metadata.impl.ZKMetadataStore#internalStoreXXX for reference.

[shibd]

About the solution, I still have some concerns.

In my understanding, the org.apache.pulsar.metadata.cache.impl.MetadataCacheImpl#refresh should be always called in the "metadata-store" Thread. Something else must be wrong with callback handling. It does not go through org.apache.pulsar.metadata.impl.AbstractMetadataStore#execute. We should fix that instead of adding synchronized for refresh.

You are right. I will solve it according to this idea and resubmit the PR.