diff --git a/.github/workflows/maven.yaml b/.github/workflows/maven.yaml index 54368e245a4..76df8e5de8a 100644 --- a/.github/workflows/maven.yaml +++ b/.github/workflows/maven.yaml @@ -70,7 +70,7 @@ jobs: profile: - {name: 'unit-tests', javaver: 11, args: 'verify -PskipQA -DskipTests=false'} - {name: 'qa-checks', javaver: 11, args: 'verify javadoc:jar -Psec-bugs -DskipTests -Dspotbugs.timeout=3600000'} - - {name: 'compat', javaver: 11, args: 'package -DskipTests -Dhadoop.version=3.0.3 -Dzookeeper.version=3.5.10'} + - {name: 'compat', javaver: 11, args: 'package -DskipTests -Dhadoop.version=3.0.3 -Dzookeeper.version=3.6.4'} - {name: 'errorprone', javaver: 11, args: 'verify -Perrorprone,skipQA'} - {name: 'jdk17', javaver: 17, args: 'verify -DskipITs'} fail-fast: false diff --git a/core/src/main/java/org/apache/accumulo/core/Constants.java b/core/src/main/java/org/apache/accumulo/core/Constants.java index 5d37166538e..075d61c9a7c 100644 --- a/core/src/main/java/org/apache/accumulo/core/Constants.java +++ b/core/src/main/java/org/apache/accumulo/core/Constants.java @@ -47,6 +47,8 @@ public class Constants { public static final String ZTABLE_COMPACT_CANCEL_ID = "/compact-cancel-id"; public static final String ZTABLE_NAMESPACE = "/namespace"; + public static final String ZTABLET_CACHE = "/tablet_cache"; + public static final String ZNAMESPACES = "/namespaces"; public static final String ZNAMESPACE_NAME = "/name"; diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java index d1a83772cdd..c2b24f4bffb 100644 --- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java +++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java @@ -520,7 +520,7 @@ private void setLocationOnce(String val, String qual, LocationType lt) { } @VisibleForTesting - static TabletMetadata create(String id, String prevEndRow, String endRow) { + public static TabletMetadata create(String id, String prevEndRow, String endRow) { TabletMetadata te = new TabletMetadata(); te.tableId = TableId.of(id); te.sawPrevEndRow = true; diff --git a/pom.xml b/pom.xml index d3ebe9291a9..fa0e49bf5d8 100644 --- a/pom.xml +++ b/pom.xml @@ -119,7 +119,7 @@ 1.70 - 5.3.0 + 5.4.0 ${project.parent.basedir}/contrib/Eclipse-Accumulo-Codestyle.xml 2.15.0 @@ -545,6 +545,14 @@ ch.qos.logback logback-core + + log4j + log4j + + + org.slf4j + slf4j-log4j12 + diff --git a/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java b/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java index 90cb171b884..5f12faf5bce 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java +++ b/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java @@ -53,6 +53,7 @@ import org.apache.accumulo.core.fate.zookeeper.ZooReader; import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter; import org.apache.accumulo.core.metadata.schema.Ample; +import org.apache.accumulo.core.metadata.schema.AmpleImpl; import org.apache.accumulo.core.rpc.SslConnectionParams; import org.apache.accumulo.core.singletons.SingletonReservation; import org.apache.accumulo.core.spi.crypto.CryptoServiceFactory; @@ -68,6 +69,7 @@ import org.apache.accumulo.server.fs.VolumeManager; import org.apache.accumulo.server.mem.LowMemoryDetector; import org.apache.accumulo.server.metadata.ServerAmpleImpl; +import org.apache.accumulo.server.metadata.TabletMetadataCache; import org.apache.accumulo.server.rpc.SaslServerConnectionParams; import org.apache.accumulo.server.rpc.ThriftServerType; import org.apache.accumulo.server.security.AuditedSecurityOperation; @@ -87,6 +89,7 @@ * and have access to the system files and configuration. */ public class ServerContext extends ClientContext { + private static final Logger log = LoggerFactory.getLogger(ServerContext.class); private final ServerInfo info; @@ -104,6 +107,9 @@ public class ServerContext extends ClientContext { private final Supplier securityOperation; private final Supplier cryptoFactorySupplier; private final Supplier lowMemoryDetector; + private final Supplier tabletMetadataCache; + + private volatile boolean tabletMetadataCacheInitialized = false; public ServerContext(SiteConfiguration siteConfig) { this(new ServerInfo(siteConfig)); @@ -130,6 +136,8 @@ private ServerContext(ServerInfo info) { memoize(() -> new AuditedSecurityOperation(this, SecurityOperation.getAuthorizor(this), SecurityOperation.getAuthenticator(this), SecurityOperation.getPermHandler(this))); lowMemoryDetector = memoize(() -> new LowMemoryDetector()); + tabletMetadataCache = memoize( + () -> new TabletMetadataCache(info.getInstanceID(), zooReaderWriter, new AmpleImpl(this))); } /** @@ -460,4 +468,24 @@ public LowMemoryDetector getLowMemoryDetector() { return lowMemoryDetector.get(); } + public TabletMetadataCache getTabletMetadataCache() { + try { + return tabletMetadataCache.get(); + } finally { + tabletMetadataCacheInitialized = true; + } + } + + public boolean isTabletMetadataCacheInitialized() { + return tabletMetadataCacheInitialized; + } + + @Override + public synchronized void close() { + super.close(); + if (tabletMetadataCacheInitialized) { + this.tabletMetadataCache.get().close(); + } + } + } diff --git a/server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java b/server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java index a939d237c14..5e68d019ec5 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java +++ b/server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java @@ -162,6 +162,8 @@ void initialize(final ServerContext context, final boolean clearInstanceName, ZooUtil.NodeExistsPolicy.FAIL); zoo.putPersistentData(zkInstanceRoot + Constants.ZSSERVERS, EMPTY_BYTE_ARRAY, ZooUtil.NodeExistsPolicy.FAIL); + zoo.putPersistentData(zkInstanceRoot + Constants.ZTABLET_CACHE, EMPTY_BYTE_ARRAY, + ZooUtil.NodeExistsPolicy.FAIL); } /** diff --git a/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMetadataCache.java b/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMetadataCache.java new file mode 100644 index 00000000000..13d68e4f12d --- /dev/null +++ b/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMetadataCache.java @@ -0,0 +1,285 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.server.metadata; + +import static java.nio.charset.StandardCharsets.UTF_8; + +import java.io.Closeable; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.accumulo.core.Constants; +import org.apache.accumulo.core.clientImpl.AcceptableThriftTableOperationException; +import org.apache.accumulo.core.data.InstanceId; +import org.apache.accumulo.core.data.TableId; +import org.apache.accumulo.core.dataImpl.KeyExtent; +import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter; +import org.apache.accumulo.core.metadata.schema.Ample; +import org.apache.accumulo.core.metadata.schema.TabletMetadata; +import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType; +import org.apache.accumulo.server.ServerContext; +import org.apache.hadoop.io.Text; +import org.apache.zookeeper.AddWatchMode; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.Watcher.WatcherType; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.ZooKeeper.States; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.LoadingCache; +import com.github.benmanes.caffeine.cache.Scheduler; + +/** + * Object that uses ZooKeeper for signaling Tablet metadata changes to keep a local cache of Tablet + * metadata up to date. + */ +public class TabletMetadataCache implements Closeable { + + private static final Logger LOG = LoggerFactory.getLogger(TabletMetadataCache.class); + private static final Long INITIAL_VALUE = Long.valueOf(0); + + private static String getWatcherPath(InstanceId iid) { + return Constants.ZROOT + "/" + iid + Constants.ZTABLET_CACHE; + } + + private static String getZNodePath(InstanceId iid) { + return getWatcherPath(iid) + "/"; + } + + /** + * Returns a String that represents this KeyExtent's TableId, endRow, and prevEndRow. + * TabletsSection#encodeRow(TableId, Text) does not preserve prevEndRow when serializing the + * KeyExtent. + * + * @param ke KeyExtent + * @return string representation of KeyExtent + */ + public static String serializeKeyExtent(KeyExtent ke) { + Text entry = new Text(ke.tableId().canonical()); + entry.append(new byte[] {';'}, 0, 1); + // TODO: May need to do semi-colon escaping in endRow and prevEndRow + // like KeyExtent.toString() + if (ke.endRow() != null) { + entry.append(ke.endRow().getBytes(), 0, ke.endRow().getLength()); + } + entry.append(new byte[] {';'}, 0, 1); + if (ke.prevEndRow() != null) { + entry.append(ke.prevEndRow().getBytes(), 0, ke.prevEndRow().getLength()); + } + return entry.toString(); + } + + /** + * Parses a String created by {@link #serializeKeyExtent(KeyExtent)} and returns a KeyExtent. + * + * @param serializedKeyExtent serialized KeyExtent + * @return KeyExtent + */ + private static KeyExtent deserializeKeyExtent(String serializedKeyExtent) { + String[] parts = serializedKeyExtent.split(";"); + assert (parts.length == 2 || parts.length == 3); + String tid = parts[0]; + String end = parts[1]; + String prev = parts.length == 3 ? parts[2] : null; + return new KeyExtent(TableId.of(tid), end == null ? null : new Text(end), + prev == null ? null : new Text(prev)); + } + + private static Long deserializeData(byte[] value) { + return Long.parseLong(new String(value, UTF_8)); + } + + private static byte[] serializeData(Long value) { + return value.toString().getBytes(UTF_8); + } + + /** + * Return the tablet_cache znode path for a KeyExtent + * + * @param extent KeyExtent + * @return tablet_cache znode path + */ + public static String getPath(InstanceId iid, KeyExtent extent) { + return getZNodePath(iid) + serializeKeyExtent(extent); + } + + /** + * Updates the data of the znode for this extent which will trigger TabletMetadataCache watchers + * to reload the TabletMetadata for this extent. + * + * @param extent KeyExtent + */ + public static void tabletMetadataChanged(ServerContext ctx, KeyExtent extent) { + final String path = getPath(ctx.getInstanceID(), extent); + try { + // remove entry from local cache only if the local + // TabletMetadataCache has been initialized + if (ctx.isTabletMetadataCacheInitialized()) { + LOG.info("Invalidating extent due to local tablet change: {}", extent); + ctx.getTabletMetadataCache().remove(extent); + } + // mutate ZK entry for this tablet + ctx.getZooReaderWriter().mutateOrCreate(path, serializeData(INITIAL_VALUE), (currVal) -> { + return serializeData(deserializeData(currVal) + 1); + }); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Interrupted updating ZooKeeper at: " + path, e); + } catch (AcceptableThriftTableOperationException | KeeperException e) { + throw new RuntimeException("Error updating ZooKeeper at: " + path, e); + } + } + + private Watcher tabletCacheZNodeWatcher = new Watcher() { + @SuppressWarnings("deprecation") + @Override + public void process(WatchedEvent event) { + LOG.trace("Event received: {}", event); + switch (event.getState()) { + case Expired: + case Disconnected: + case AuthFailed: + case Closed: + case ConnectedReadOnly: + // Connection issue, invalidate all and stop caching + // until we are connected again. + if (connected.compareAndSet(true, false)) { + LOG.info("Connection issue, clearing cache."); + tabletMetadataCache.invalidateAll(); + } + break; + case SyncConnected: + // Connected, begin caching + if (connected.compareAndSet(false, true)) { + LOG.info("Connection established."); + } + break; + case NoSyncConnected: + case SaslAuthenticated: + case Unknown: + default: + // No ops + break; + + } + switch (event.getType()) { + case NodeCreated: + // Do nothing, cache will be populated on clients first call to get() + break; + case NodeDataChanged: + case NodeDeleted: + // Remove the tablet metadata cache entry on update or delete. + KeyExtent extent = getExtent(event.getPath()); + LOG.info("Invalidating extent: {}", extent); + tabletMetadataCache.invalidate(extent); + break; + case None: + case NodeChildrenChanged: // These events are not triggered by persistent recursive watches + case ChildWatchRemoved: + case DataWatchRemoved: + case PersistentWatchRemoved: + default: + break; + + } + } + }; + + private final String watcherPath; + private final String znodePath; + private final Ample ample; + private final ZooKeeper zoo; + protected final AtomicBoolean connected = new AtomicBoolean(false); + protected final LoadingCache tabletMetadataCache; + + public TabletMetadataCache(final InstanceId iid, final ZooReaderWriter zrw, final Ample ample) { + + this.zoo = zrw.getZooKeeper(); + this.ample = ample; + watcherPath = getWatcherPath(iid); + znodePath = getZNodePath(iid); + + // TODO: Likely want to set a max size and eviction parameters + tabletMetadataCache = + Caffeine.newBuilder().recordStats().scheduler(Scheduler.systemScheduler()).build((k) -> { + TabletMetadata tm = ample.readTablet(k, ColumnType.values()); + LOG.debug("Loading tablet metadata for extent: {}. Returned: {}", k, tm); + return tm; + }); + + try { + zoo.addWatch(watcherPath, tabletCacheZNodeWatcher, AddWatchMode.PERSISTENT_RECURSIVE); + } catch (KeeperException e) { + throw new RuntimeException("Error setting watch", e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Thread interrupted setting watch", e); + } + + } + + /** + * Return a KeyExtent object from the tablet_cache znode entry + * + * @param path tablet_cache znode path + * @return KeyExtent object + */ + protected KeyExtent getExtent(String path) { + return deserializeKeyExtent(path.substring(znodePath.length())); + } + + /** + * Returns cached TabletMetadata for the KeyExtent. If it does not exist, then it will be loaded + * into the cache via the CacheLoader which uses Ample to read the TabletMetadata and all of its + * columns. If the ZooKeeper connection is down, this will read from ample and skip the cache. + * + * @param extent KeyExtent + * @return TabletMetadata for the KeyExtent + */ + public TabletMetadata get(KeyExtent extent) { + if (!States.CONNECTED.equals(zoo.getState()) || !connected.get()) { + return ample.readTablet(extent, ColumnType.values()); + } + return tabletMetadataCache.get(extent); + } + + /** + * Remove the cache entry for this KeyExtent + * + * @param extent KeyExtent + */ + private void remove(KeyExtent extent) { + tabletMetadataCache.invalidate(extent); + } + + @Override + public void close() { + tabletMetadataCache.invalidateAll(); + tabletMetadataCache.cleanUp(); + try { + zoo.removeWatches(watcherPath, tabletCacheZNodeWatcher, WatcherType.Any, false); + } catch (InterruptedException | KeeperException e) { + LOG.error("Error removing persistent watcher", e); + } + } + +} diff --git a/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMutatorBase.java b/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMutatorBase.java index 62d13017b49..f252fa596e3 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMutatorBase.java +++ b/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMutatorBase.java @@ -56,13 +56,15 @@ public abstract class TabletMutatorBase implements Ample.TabletMutator { - private final ServerContext context; + protected final ServerContext context; + protected final KeyExtent extent; private final Mutation mutation; protected AutoCloseable closeAfterMutate; private boolean updatesEnabled = true; protected TabletMutatorBase(ServerContext context, KeyExtent extent) { this.context = context; + this.extent = extent; mutation = new Mutation(extent.toMetaRow()); } diff --git a/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMutatorImpl.java b/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMutatorImpl.java index e4889871da8..896c474d82a 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMutatorImpl.java +++ b/server/base/src/main/java/org/apache/accumulo/server/metadata/TabletMutatorImpl.java @@ -43,6 +43,7 @@ public void mutate() { } catch (Exception e) { throw new RuntimeException(e); } + TabletMetadataCache.tabletMetadataChanged(context, extent); } } diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java index 33c0a415a5a..d93d06894d7 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java +++ b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java @@ -96,6 +96,7 @@ import org.apache.accumulo.core.util.Pair; import org.apache.accumulo.server.ServerContext; import org.apache.accumulo.server.gc.AllVolumesDirectory; +import org.apache.accumulo.server.metadata.TabletMetadataCache; import org.apache.hadoop.io.Text; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -157,6 +158,9 @@ public static void update(ServerContext context, Writer t, ServiceLock zooLock, while (true) { try { t.update(m); + if (!extent.isMeta()) { + TabletMetadataCache.tabletMetadataChanged(context, extent); + } return; } catch (AccumuloException | TableNotFoundException | AccumuloSecurityException e) { logUpdateFailure(m, extent, e); diff --git a/test/src/main/java/org/apache/accumulo/test/functional/TabletMetadataCacheIT.java b/test/src/main/java/org/apache/accumulo/test/functional/TabletMetadataCacheIT.java new file mode 100644 index 00000000000..3b32108283b --- /dev/null +++ b/test/src/main/java/org/apache/accumulo/test/functional/TabletMetadataCacheIT.java @@ -0,0 +1,352 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.test.functional; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.apache.accumulo.harness.AccumuloITBase.ZOOKEEPER_TESTING_SERVER; +import static org.easymock.EasyMock.expect; +import static org.easymock.EasyMock.replay; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.File; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.accumulo.core.Constants; +import org.apache.accumulo.core.clientImpl.AcceptableThriftTableOperationException; +import org.apache.accumulo.core.data.InstanceId; +import org.apache.accumulo.core.data.TableId; +import org.apache.accumulo.core.dataImpl.KeyExtent; +import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter; +import org.apache.accumulo.core.fate.zookeeper.ZooUtil; +import org.apache.accumulo.core.metadata.schema.Ample; +import org.apache.accumulo.core.metadata.schema.AmpleImpl; +import org.apache.accumulo.core.metadata.schema.TabletMetadata; +import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType; +import org.apache.accumulo.server.ServerContext; +import org.apache.accumulo.server.metadata.TabletMetadataCache; +import org.apache.accumulo.test.zookeeper.ZooKeeperTestingServer; +import org.apache.hadoop.io.Text; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.ZooKeeper; +import org.easymock.EasyMock; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.MethodOrderer; +import org.junit.jupiter.api.Order; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestMethodOrder; +import org.junit.jupiter.api.io.TempDir; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.github.benmanes.caffeine.cache.stats.CacheStats; + +@Tag(ZOOKEEPER_TESTING_SERVER) +@TestMethodOrder(MethodOrderer.OrderAnnotation.class) +public class TabletMetadataCacheIT { + + /** + * Class for test that exists to expose protected methods for test + */ + public static class TestTabletMetadataCache extends TabletMetadataCache { + + public TestTabletMetadataCache(InstanceId iid, ZooReaderWriter zrw, Ample ample) { + super(iid, zrw, ample); + } + + @Override + public KeyExtent getExtent(String path) { + return super.getExtent(path); + } + + public AtomicBoolean getConnected() { + return connected; + } + + /** + * Return size of the cache + * + * @return size + */ + protected int getTabletMetadataCacheSize() { + return tabletMetadataCache.asMap().size(); + } + + /** + * Return Cache stats object + * + * @return cache stats + */ + protected CacheStats getTabletMetadataCacheStats() { + return tabletMetadataCache.stats(); + } + + } + + @TempDir + private static File tempDir; + + private static final InstanceId IID = InstanceId.of(UUID.randomUUID()); + + private static ZooKeeperTestingServer szk = null; + private static ZooKeeper zooKeeper; + + private ServerContext context = null; + private TestTabletMetadataCache cache = null; + private TableId tid = TableId.of("2"); + private KeyExtent ke1 = new KeyExtent(tid, new Text("b"), null); + private KeyExtent ke2 = new KeyExtent(tid, new Text("d"), new Text("b")); + private KeyExtent ke3 = new KeyExtent(tid, new Text("g"), new Text("d")); + private KeyExtent ke4 = new KeyExtent(tid, new Text("m"), new Text("g")); + private TabletMetadata tm1 = TabletMetadata.create(tid.canonical(), null, "b"); + private TabletMetadata tm2 = TabletMetadata.create(tid.canonical(), "b", "d"); + private TabletMetadata tm3 = TabletMetadata.create(tid.canonical(), "d", "g"); + private TabletMetadata tm4 = TabletMetadata.create(tid.canonical(), "g", "m"); + + @BeforeAll + public static void setup() throws Exception { + szk = new ZooKeeperTestingServer(tempDir); + szk.initPaths(ZooUtil.getRoot(IID) + Constants.ZTABLET_CACHE); + zooKeeper = szk.getZooKeeper(); + ZooUtil.digestAuth(zooKeeper, ZooKeeperTestingServer.SECRET); + } + + @AfterAll + public static void teardown() throws Exception { + szk.close(); + } + + @BeforeEach + public void before() { + context = EasyMock.createNiceMock(ServerContext.class); + AmpleImpl ample = EasyMock.createStrictMock(AmpleImpl.class); + expect(ample.readTablet(ke1, ColumnType.values())).andReturn(tm1).anyTimes(); + expect(ample.readTablet(ke2, ColumnType.values())).andReturn(tm2).anyTimes(); + expect(ample.readTablet(ke3, ColumnType.values())).andReturn(tm3).anyTimes(); + expect(ample.readTablet(ke4, ColumnType.values())).andReturn(tm4).anyTimes(); + cache = new TestTabletMetadataCache(IID, szk.getZooReaderWriter(), ample); + expect(context.getInstanceID()).andReturn(IID).anyTimes(); + expect(context.getZooReaderWriter()).andReturn(szk.getZooReaderWriter()).anyTimes(); + expect(context.getAmple()).andReturn(ample).anyTimes(); + expect(context.getZooKeepers()).andReturn(szk.getConn()); + expect(context.getZooKeepersSessionTimeOut()).andReturn((30000)); + expect(context.isTabletMetadataCacheInitialized()).andReturn(Boolean.TRUE).anyTimes(); + expect(context.getTabletMetadataCache()).andReturn(cache).anyTimes(); + replay(context, ample); + } + + @AfterEach + public void after() { + cache.close(); + } + + @Test + @Order(1) + public void testPathParsing() { + String path = TabletMetadataCache.getPath(IID, ke1); + assertEquals("/accumulo/" + IID + Constants.ZTABLET_CACHE + "/2;b;", path); + KeyExtent result = cache.getExtent(path); + assertEquals(ke1, result); + String path2 = TabletMetadataCache.getPath(IID, ke2); + assertEquals("/accumulo/" + IID + Constants.ZTABLET_CACHE + "/2;d;b", path2); + KeyExtent result2 = cache.getExtent(path2); + assertEquals(ke2.toMetaRow(), result2.toMetaRow()); + } + + @Test + @Order(2) + public void testCachingLocalTabletChange() { + TabletMetadataCache.tabletMetadataChanged(context, ke1); // This will perform a create in ZK + assertEquals(0, cache.getTabletMetadataCacheStats().loadCount()); + assertEquals(0, cache.getTabletMetadataCacheSize()); + TabletMetadata result = cache.get(ke1); + assertEquals(1, cache.getTabletMetadataCacheSize()); + assertEquals(1, cache.getTabletMetadataCacheStats().requestCount()); + assertEquals(1, cache.getTabletMetadataCacheStats().loadSuccessCount()); + assertEquals(tm1, result); + TabletMetadataCache.tabletMetadataChanged(context, ke1); // This will perform an update and + // trigger invalidation + assertEquals(0, cache.getTabletMetadataCacheSize()); + result = cache.get(ke1); + assertEquals(1, cache.getTabletMetadataCacheSize()); + assertEquals(2, cache.getTabletMetadataCacheStats().requestCount()); + assertEquals(2, cache.getTabletMetadataCacheStats().loadSuccessCount()); + assertEquals(tm1, result); + } + + public class TabletChangedThread implements Runnable { + + private final Logger LOG = LoggerFactory.getLogger(TabletChangedThread.class); + + private Long deserialize(byte[] value) { + return Long.parseLong(new String(value, UTF_8)); + } + + private byte[] serialize(Long value) { + return value.toString().getBytes(UTF_8); + } + + @Override + public void run() { + + // mutate ZK entry for this tablet + String path = Constants.ZROOT + "/" + IID + Constants.ZTABLET_CACHE + "/" + + TabletMetadataCache.serializeKeyExtent(ke2); + try { + LOG.info("Mutating node at path: {}", path); + szk.getZooReaderWriter().mutateOrCreate(path, serialize(Long.valueOf(0)), (currVal) -> { + return serialize(deserialize(currVal) + 1); + }); + } catch (AcceptableThriftTableOperationException | KeeperException | InterruptedException e) { + LOG.error("***ERROR***", e); + } + } + } + + @Test + @Order(3) + public void testCachingThreadTabletChange() throws InterruptedException { + TabletMetadataCache.tabletMetadataChanged(context, ke2); // This will perform a create in ZK + + assertEquals(0, cache.getTabletMetadataCacheStats().loadCount()); + assertEquals(0, cache.getTabletMetadataCacheSize()); + TabletMetadata result = cache.get(ke2); + assertEquals(1, cache.getTabletMetadataCacheSize()); + assertEquals(1, cache.getTabletMetadataCacheStats().requestCount()); + assertEquals(1, cache.getTabletMetadataCacheStats().loadSuccessCount()); + assertEquals(tm2, result); + + Thread t = new Thread(new TabletChangedThread()); + t.start(); + t.join(); + + // wait for zk watcher to remove cache entry + while (cache.getTabletMetadataCacheSize() != 0) { + Thread.sleep(100); + } + + assertEquals(0, cache.getTabletMetadataCacheSize()); + result = cache.get(ke2); + assertEquals(1, cache.getTabletMetadataCacheSize()); + assertEquals(2, cache.getTabletMetadataCacheStats().requestCount()); + assertEquals(2, cache.getTabletMetadataCacheStats().loadSuccessCount()); + assertEquals(tm2, result); + } + + @Test + @Order(4) + public void testZooKeeperConnectionRestart() throws Exception { + TabletMetadataCache.tabletMetadataChanged(context, ke1); // This will perform a create in ZK + assertEquals(0, cache.getTabletMetadataCacheStats().loadCount()); + assertEquals(0, cache.getTabletMetadataCacheSize()); + TabletMetadata result = cache.get(ke1); + assertEquals(1, cache.getTabletMetadataCacheSize()); + assertEquals(1, cache.getTabletMetadataCacheStats().requestCount()); + assertEquals(1, cache.getTabletMetadataCacheStats().loadSuccessCount()); + assertEquals(tm1, result); + TabletMetadataCache.tabletMetadataChanged(context, ke1); // This will perform an update and + // trigger invalidation + assertEquals(0, cache.getTabletMetadataCacheSize()); + result = cache.get(ke1); + assertEquals(1, cache.getTabletMetadataCacheSize()); + assertEquals(2, cache.getTabletMetadataCacheStats().requestCount()); + assertEquals(2, cache.getTabletMetadataCacheStats().loadSuccessCount()); + assertEquals(tm1, result); + + // Close ZK Server + szk.close(); + LoggerFactory.getLogger(TabletMetadataCacheIT.class).info("ZK Server closed"); + + long requestCount = 0L; + // Wait to be disconnected + while (cache.getConnected().get()) { + // TODO: The ZooKeeper connection does not realize that it's disconnected + // immediately. Depending on the type of connection failure there is some + // amount of time before the Watcher gets notified. For example, it appears + // that if the server closes, like in this test, then the Watcher gets notified + // rather quickly (less than 100ms). In the case where there is a network partition + // or the ZK Server is restarted, the Watcher might not get the event for some time. + // This means that because the ZooKeeper client does not know immediately that it's + // disconnected, that it's possible that the cache will return a possibly stale result + result = cache.get(ke1); + requestCount = cache.getTabletMetadataCacheStats().requestCount(); + assertEquals(1, cache.getTabletMetadataCacheSize()); + assertTrue(requestCount > 2); + assertEquals(3, cache.getTabletMetadataCacheStats().loadSuccessCount()); + assertEquals(tm1, result); + Thread.sleep(10); + } + + // Start ZK Server + szk.restart(); + + // Wait to be reconnected + while (!cache.getConnected().get()) { + Thread.sleep(10); + } + + assertEquals(0, cache.getTabletMetadataCacheSize()); + result = cache.get(ke1); + assertEquals(1, cache.getTabletMetadataCacheSize()); + assertEquals(requestCount + 1, cache.getTabletMetadataCacheStats().requestCount()); + assertEquals(4, cache.getTabletMetadataCacheStats().loadSuccessCount()); + assertEquals(tm1, result); + + } + + @Test + @Order(5) + public void testZooKeeperDies() throws Exception { + TabletMetadataCache.tabletMetadataChanged(context, ke1); // This will perform a create in ZK + assertEquals(0, cache.getTabletMetadataCacheStats().loadCount()); + assertEquals(0, cache.getTabletMetadataCacheSize()); + TabletMetadata result = cache.get(ke1); + assertEquals(1, cache.getTabletMetadataCacheSize()); + assertEquals(1, cache.getTabletMetadataCacheStats().requestCount()); + assertEquals(1, cache.getTabletMetadataCacheStats().loadSuccessCount()); + assertEquals(tm1, result); + TabletMetadataCache.tabletMetadataChanged(context, ke1); // This will perform an update and + // trigger invalidation + assertEquals(0, cache.getTabletMetadataCacheSize()); + result = cache.get(ke1); + assertEquals(1, cache.getTabletMetadataCacheSize()); + assertEquals(2, cache.getTabletMetadataCacheStats().requestCount()); + assertEquals(2, cache.getTabletMetadataCacheStats().loadSuccessCount()); + assertEquals(tm1, result); + + // close ZooKeeper + szk.close(); + + // Wait to be disconnected + while (cache.getConnected().get()) { + Thread.sleep(10); + } + + assertEquals(0, cache.getTabletMetadataCacheSize()); + result = cache.get(ke1); + assertEquals(0, cache.getTabletMetadataCacheSize()); + + } + +} diff --git a/test/src/main/java/org/apache/accumulo/test/zookeeper/ZooKeeperTestingServer.java b/test/src/main/java/org/apache/accumulo/test/zookeeper/ZooKeeperTestingServer.java index dd83d8cf4b6..dd894e1390c 100644 --- a/test/src/main/java/org/apache/accumulo/test/zookeeper/ZooKeeperTestingServer.java +++ b/test/src/main/java/org/apache/accumulo/test/zookeeper/ZooKeeperTestingServer.java @@ -130,6 +130,12 @@ public void initPaths(String s) { } } + public void restart() throws Exception { + if (zkServer != null) { + zkServer.restart(); + } + } + @Override public void close() throws IOException { if (zkServer != null) {