[metadata-store] Fix metadata cache inconsistency on doing refresh

[Demogorgon314]

Fixes #14110 #14096 #14219

Motivation

Currently, the metadata cache refresh has race condition.

The main reason is MetadataCacheImpl#refresh is not atomic operations.

// 1. Caffeine.asyncLoad already begin and reads an old value but is not done yet.
if (objCache.getIfPresent(path) != null) {
  objCache.synchronous().invalidate(path);
  // 2. Caffeine.asyncLoad done, the old value will store to cache.
  objCache.synchronous().refresh(path);
}

I added some log in MetadataCacheImpl in order to observe:

@Override
public CompletableFuture<Optional<CacheGetResult<T>>> asyncLoad(String key, Executor executor) {
  log.info("Async load for objCache {}", key);
  CompletableFuture<Optional<CacheGetResult<T>>> future = readValueFromStore(key);
  future.whenComplete((op, ex) -> {
    if (ex == null){
      op.ifPresent(tCacheGetResult -> log.info("Async load for objCache complete {} - {}", key, tCacheGetResult));
    }
  });
  return future;
}
---------------------
@Override
public void refresh(String path) {
  log.info("1. Refresh path {}", path);
  // Refresh object of path if only it is cached before.
  if (objCache.getIfPresent(path) != null) {
    objCache.synchronous().invalidate(path);
    objCache.synchronous().refresh(path);
    log.info("2. Refresh path complete {}:{}", path, get(path).join().get());
  } else {
    log.info("0. Skip Refresh path {}", path);
  }
}

Cache inconsistency, you can see the async load for objCache complete in mid of refresh :

2022-02-15T08:33:06,669+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@272 - 1. Refresh path /key-33552641536875 {}
2022-02-15T08:33:06,669+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@279 - 0. Skip Refresh path /key-33552641536875 {}
2022-02-15T08:33:06,669+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@80 - Async load for objCache /key-33552641536875 {}
2022-02-15T08:33:06,669+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@84 - Async load for objCache complete /key-33552641536875 - CacheGetResult(value=MetadataCacheTest.MyClass(a=a, b=1), stat=Stat(path=/key-33552641536875, version=0, creationTimestamp=1644885186668, modificationTimestamp=1644885186668, ephemeral=false, createdBySelf=true)) {}
2022-02-15T08:33:06,669+0800 INFO  [metadata-store-2437-1] o.a.p.m.c.i.MetadataCacheImpl@272 - 1. Refresh path /key-33552641536875 {}
2022-02-15T08:33:06,669+0800 INFO  [metadata-store-2437-1] o.a.p.m.c.i.MetadataCacheImpl@80 - Async load for objCache /key-33552641536875 {}
2022-02-15T08:33:06,670+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@272 - 1. Refresh path /key-33552641536875 {}
2022-02-15T08:33:06,670+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@279 - 0. Skip Refresh path /key-33552641536875 {}
2022-02-15T08:33:06,670+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@272 - 1. Refresh path /key-33552641536875 {}
2022-02-15T08:33:06,670+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@279 - 0. Skip Refresh path /key-33552641536875 {}
2022-02-15T08:33:06,671+0800 INFO  [metadata-store-2437-1] o.a.p.m.c.i.MetadataCacheImpl@84 - Async load for objCache complete /key-33552641536875 - CacheGetResult(value=MetadataCacheTest.MyClass(a=a, b=1), stat=Stat(path=/key-33552641536875, version=0, creationTimestamp=1644885186668, modificationTimestamp=1644885186668, ephemeral=false, createdBySelf=true)) {}
2022-02-15T08:33:06,671+0800 INFO  [metadata-store-2437-1] o.a.p.m.c.i.MetadataCacheImpl@277 - 2. Refresh path complete /key-33552641536875:MetadataCacheTest.MyClass(a=a, b=1) {}
2022-02-15T08:33:06,671+0800 INFO  [metadata-store-2437-1] o.a.p.m.c.i.MetadataCacheImpl@272 - 1. Refresh path /key-33552641536875 {}
2022-02-15T08:33:06,671+0800 INFO  [metadata-store-2437-1] o.a.p.m.c.i.MetadataCacheImpl@80 - Async load for objCache /key-33552641536875 {}
2022-02-15T08:33:06,671+0800 INFO  [metadata-store-2437-1] o.a.p.m.c.i.MetadataCacheImpl@84 - Async load for objCache complete /key-33552641536875 - CacheGetResult(value=MetadataCacheTest.MyClass(a=a, b=2), stat=Stat(path=/key-33552641536875, version=1, creationTimestamp=1644885186668, modificationTimestamp=1644885186669, ephemeral=false, createdBySelf=true)) {}
2022-02-15T08:33:06,671+0800 INFO  [metadata-store-2437-1] o.a.p.m.c.i.MetadataCacheImpl@277 - 2. Refresh path complete /key-33552641536875:MetadataCacheTest.MyClass(a=a, b=2) {}

Correct log:

2022-02-15T08:33:06,659+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@272 - 1. Refresh path /key-33552632155000 {}
2022-02-15T08:33:06,659+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@279 - 0. Skip Refresh path /key-33552632155000 {}
2022-02-15T08:33:06,659+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@80 - Async load for objCache /key-33552632155000 {}
2022-02-15T08:33:06,659+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@84 - Async load for objCache complete /key-33552632155000 - CacheGetResult(value=MetadataCacheTest.MyClass(a=a, b=1), stat=Stat(path=/key-33552632155000, version=0, creationTimestamp=1644885186659, modificationTimestamp=1644885186659, ephemeral=false, createdBySelf=true)) {}
2022-02-15T08:33:06,661+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@272 - 1. Refresh path /key-33552632155000 {}
2022-02-15T08:33:06,661+0800 INFO  [metadata-store-2436-1] o.a.p.m.c.i.MetadataCacheImpl@272 - 1. Refresh path /key-33552632155000 {}
2022-02-15T08:33:06,661+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@80 - Async load for objCache /key-33552632155000 {}
2022-02-15T08:33:06,661+0800 INFO  [metadata-store-2436-1] o.a.p.m.c.i.MetadataCacheImpl@279 - 0. Skip Refresh path /key-33552632155000 {}
2022-02-15T08:33:06,661+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@84 - Async load for objCache complete /key-33552632155000 - CacheGetResult(value=MetadataCacheTest.MyClass(a=a, b=2), stat=Stat(path=/key-33552632155000, version=1, creationTimestamp=1644885186659, modificationTimestamp=1644885186659, ephemeral=false, createdBySelf=true)) {}
2022-02-15T08:33:06,661+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@277 - 2. Refresh path complete /key-33552632155000:MetadataCacheTest.MyClass(a=a, b=2) {}
2022-02-15T08:33:06,661+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@272 - 1. Refresh path /key-33552632155000 {}
2022-02-15T08:33:06,661+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@80 - Async load for objCache /key-33552632155000 {}
2022-02-15T08:33:06,661+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@84 - Async load for objCache complete /key-33552632155000 - CacheGetResult(value=MetadataCacheTest.MyClass(a=a, b=2), stat=Stat(path=/key-33552632155000, version=1, creationTimestamp=1644885186659, modificationTimestamp=1644885186659, ephemeral=false, createdBySelf=true)) {}
2022-02-15T08:33:06,661+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@277 - 2. Refresh path complete /key-33552632155000:MetadataCacheTest.MyClass(a=a, b=2) {}
2022-02-15T08:33:06,662+0800 INFO  [main] o.a.p.m.c.i.MetadataCacheImpl@80 - Async load for objCache /key-33552635148375 {}

Modifications

Make metadata cache refresh method atomic.

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

Need to update docs?

• no-need-doc

[gaozhangmin]

@Demogorgon314 I still get test error: https://github.com/apache/pulsar/runs/5231565908?check_suite_focus=true

[Demogorgon314]

@Demogorgon314 I still get test error: https://github.com/apache/pulsar/runs/5231565908?check_suite_focus=true

I'll push a PR to fix it.

[Demogorgon314]

@gaozhangmin PTAL #14354 : )