Skip to content

Commit

Permalink
Revert "Add latch for ZK hydration initialization (#655)"
Browse files Browse the repository at this point in the history
This reverts commit 8d8024c.
  • Loading branch information
bryanlb authored Sep 1, 2023
1 parent cad83fb commit daa5c85
Show file tree
Hide file tree
Showing 2 changed files with 4 additions and 89 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,12 @@

import static com.slack.kaldb.server.KaldbConfig.DEFAULT_ZK_TIMEOUT_SECS;

import com.slack.kaldb.util.RuntimeHalterImpl;
import java.io.Closeable;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
Expand All @@ -35,8 +33,6 @@ public class KaldbMetadataStore<T extends KaldbMetadata> implements Closeable {

private final ZPath zPath;

private final CountDownLatch cacheInitialized = new CountDownLatch(1);

protected final ModeledFramework<T> modeledClient;

private final CachedModeledFramework<T> cachedModeledFramework;
Expand Down Expand Up @@ -65,7 +61,6 @@ public KaldbMetadataStore(

if (shouldCache) {
cachedModeledFramework = modeledClient.cached();
cachedModeledFramework.listenable().addListener(getCacheInitializedListener());
cachedModeledFramework.start();
} else {
cachedModeledFramework = null;
Expand Down Expand Up @@ -104,7 +99,6 @@ public T getSync(String path) {

public CompletionStage<Stat> hasAsync(String path) {
if (cachedModeledFramework != null) {
awaitCacheInitialized();
return cachedModeledFramework.withPath(zPath.resolved(path)).checkExists();
}
return modeledClient.withPath(zPath.resolved(path)).checkExists();
Expand Down Expand Up @@ -161,11 +155,9 @@ public void deleteSync(T metadataNode) {
}

public CompletionStage<List<T>> listAsync() {
if (cachedModeledFramework == null) {
if (cachedModeledFramework == null)
throw new UnsupportedOperationException("Caching is disabled");
}

awaitCacheInitialized();
return cachedModeledFramework.list();
}

Expand All @@ -178,9 +170,8 @@ public List<T> listSync() {
}

public void addListener(KaldbMetadataStoreChangeListener<T> watcher) {
if (cachedModeledFramework == null) {
if (cachedModeledFramework == null)
throw new UnsupportedOperationException("Caching is disabled");
}

// this mapping exists because the remove is by reference, and the listener is a different
// object type
Expand All @@ -196,35 +187,11 @@ public void addListener(KaldbMetadataStoreChangeListener<T> watcher) {
}

public void removeListener(KaldbMetadataStoreChangeListener<T> watcher) {
if (cachedModeledFramework == null) {
if (cachedModeledFramework == null)
throw new UnsupportedOperationException("Caching is disabled");
}
cachedModeledFramework.listenable().removeListener(listenerMap.remove(watcher));
}

private void awaitCacheInitialized() {
try {
cacheInitialized.await();
} catch (InterruptedException e) {
new RuntimeHalterImpl().handleFatal(e);
}
}

private ModeledCacheListener<T> getCacheInitializedListener() {
return new ModeledCacheListener<T>() {
@Override
public void accept(Type type, ZPath path, Stat stat, T model) {
// no-op
}

@Override
public void initialized() {
ModeledCacheListener.super.initialized();
cacheInitialized.countDown();
}
};
}

@Override
public synchronized void close() {
if (cachedModeledFramework != null) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ public void setUp() throws Exception {
KaldbConfigs.ZookeeperConfig.newBuilder()
.setZkConnectString(testingServer.getConnectString())
.setZkPathPrefix("Test")
.setZkSessionTimeoutMs(10000)
.setZkSessionTimeoutMs(1000)
.setZkConnectionTimeoutMs(1000)
.setSleepBetweenRetriesMs(500)
.build();
Expand Down Expand Up @@ -394,56 +394,4 @@ public TestMetadataStore() {
throw new RuntimeException(e);
}
}

@Test
public void testSlowCacheInitialization() {
class FastMetadataStore extends KaldbMetadataStore<TestMetadata> {
public FastMetadataStore() {
super(
curatorFramework,
CreateMode.PERSISTENT,
true,
new JacksonModelSerializer<>(TestMetadata.class),
STORE_FOLDER);
}
}

class SlowSerializer implements ModelSerializer<TestMetadata> {
final JacksonModelSerializer<TestMetadata> serializer =
new JacksonModelSerializer<>(TestMetadata.class);

@Override
public byte[] serialize(TestMetadata model) {
return serializer.serialize(model);
}

@Override
public TestMetadata deserialize(byte[] bytes) {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return serializer.deserialize(bytes);
}
}

class SlowMetadataStore extends KaldbMetadataStore<TestMetadata> {
public SlowMetadataStore() {
super(curatorFramework, CreateMode.PERSISTENT, true, new SlowSerializer(), STORE_FOLDER);
}
}

int testMetadataInitCount = 10;
try (KaldbMetadataStore<TestMetadata> init = new FastMetadataStore()) {
for (int i = 0; i < testMetadataInitCount; i++) {
init.createSync(new TestMetadata("name" + i, "value" + i));
}
}

try (KaldbMetadataStore<TestMetadata> init = new SlowMetadataStore()) {
List<TestMetadata> metadata = init.listSync();
assertThat(metadata.size()).isEqualTo(testMetadataInitCount);
}
}
}

0 comments on commit daa5c85

Please sign in to comment.