Skip to content

Commit

Permalink
Merge branch 'main' into common-zk-path-creation
Browse files Browse the repository at this point in the history
  • Loading branch information
dlmarion committed Jan 13, 2025
2 parents da51e06 + e745a5d commit 21bd629
Show file tree
Hide file tree
Showing 210 changed files with 3,304 additions and 3,505 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ static ContextClassLoaderFactory getContextFactory() {
}

// for testing
static synchronized void resetContextFactoryForTests() {
public static synchronized void resetContextFactoryForTests() {
FACTORY = null;
}

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,136 @@
/*
* 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.core.client.admin;

import java.io.Serializable;
import java.time.Duration;
import java.util.Objects;
import java.util.Optional;

import com.google.common.base.Preconditions;

/**
* @since 4.0.0
*/
public class TabletMergeability implements Serializable {
private static final long serialVersionUID = 1L;

private static final TabletMergeability NEVER = new TabletMergeability();
private static final TabletMergeability ALWAYS = new TabletMergeability(Duration.ZERO);

private final Duration delay;

private TabletMergeability(Duration delay) {
this.delay = Objects.requireNonNull(delay);
}

// Edge case for NEVER
private TabletMergeability() {
this.delay = null;
}

/**
* Determines if the configured delay signals a tablet is never eligible to be automatically
* merged.
*
* @return true if never mergeable, else false
*/
public boolean isNever() {
return this.delay == null;
}

/**
* Determines if the configured delay signals a tablet is always eligible to be automatically
* merged now. (Has a delay of 0)
*
* @return true if always mergeable now, else false
*/
public boolean isAlways() {
return delay != null && this.delay.isZero();
}

/**
* Returns an Optional duration of the delay which is one of:
*
* <ul>
* <li>empty (never)</li>
* <li>0 (now)</li>
* <li>positive delay</li>
* </ul>
*
* @return the configured mergeability delay
*/
public Optional<Duration> getDelay() {
return Optional.ofNullable(delay);
}

@Override
public boolean equals(Object o) {
if (o == null || getClass() != o.getClass()) {
return false;
}
TabletMergeability that = (TabletMergeability) o;
return Objects.equals(delay, that.delay);
}

@Override
public int hashCode() {
return Objects.hashCode(delay);
}

@Override
public String toString() {
if (delay == null) {
return "TabletMergeability=NEVER";
}
return "TabletMergeability=AFTER:" + delay.toMillis() + "ms";
}

/**
* Signifies that a tablet is never eligible to be automatically merged.
*
* @return a {@link TabletMergeability} with an empty delay signaling never merge
*/
public static TabletMergeability never() {
return NEVER;
}

/**
* Signifies that a tablet is eligible now to be automatically merged
*
* @return a {@link TabletMergeability} with a delay of 0 signaling never merge
*/
public static TabletMergeability always() {
return ALWAYS;
}

/**
* Creates a {@link TabletMergeability} that signals a tablet has a delay to a point in the future
* before it is automatically eligible to be merged. The duration must be positive value.
*
* @param delay the duration of the delay
*
* @return a {@link TabletMergeability} from the given delay.
*/
public static TabletMergeability after(Duration delay) {
Preconditions.checkArgument(delay.toNanos() >= 0, "Duration of delay must be greater than 0.");
return new TabletMergeability(delay);
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Suppliers.memoize;
import static com.google.common.base.Suppliers.memoizeWithExpiration;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.accumulo.core.util.threads.ThreadPoolNames.CONDITIONAL_WRITER_CLEANUP_POOL;
Expand All @@ -44,11 +43,11 @@
import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
Expand Down Expand Up @@ -79,8 +78,6 @@
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.fate.zookeeper.ZooCache;
import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZcStat;
import org.apache.accumulo.core.fate.zookeeper.ZooCacheFactory;
import org.apache.accumulo.core.fate.zookeeper.ZooReader;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
Expand All @@ -105,6 +102,7 @@
import org.apache.accumulo.core.util.tables.TableZooHelper;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.core.util.threads.Threads;
import org.apache.accumulo.core.zookeeper.ZooSession;
import org.apache.hadoop.conf.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -127,14 +125,12 @@ public class ClientContext implements AccumuloClient {
private static final Logger log = LoggerFactory.getLogger(ClientContext.class);

private final ClientInfo info;
private InstanceId instanceId;
private final ZooReader zooReader;
private final ZooCache zooCache;
private final Supplier<ZooCache> zooCache;

private Credentials creds;
private BatchWriterConfig batchWriterConfig;
private ConditionalWriterConfig conditionalWriterConfig;
private final AccumuloConfiguration serverConf;
private final AccumuloConfiguration accumuloConf;
private final Configuration hadoopConf;

// These fields are very frequently accessed (each time a connection is created) and expensive to
Expand Down Expand Up @@ -162,6 +158,9 @@ public class ClientContext implements AccumuloClient {
private MeterRegistry micrometer;
private Caches caches;

private final AtomicBoolean zooKeeperOpened = new AtomicBoolean(false);
private final Supplier<ZooSession> zooSession;

private void ensureOpen() {
if (closed) {
throw new IllegalStateException("This client was closed.");
Expand Down Expand Up @@ -228,13 +227,19 @@ public ClientContext(SingletonReservation reservation, ClientInfo info,
AccumuloConfiguration serverConf, UncaughtExceptionHandler ueh) {
this.info = info;
this.hadoopConf = info.getHadoopConf();
zooReader = new ZooReader(info.getZooKeepers(), info.getZooKeepersSessionTimeOut());
zooCache =
new ZooCacheFactory().getZooCache(info.getZooKeepers(), info.getZooKeepersSessionTimeOut());
this.serverConf = serverConf;

this.zooSession = memoize(() -> {
var zk = info
.getZooKeeperSupplier(getClass().getSimpleName() + "(" + info.getPrincipal() + ")").get();
zooKeeperOpened.set(true);
return zk;
});

this.zooCache = memoize(() -> new ZooCache(getZooSession()));
this.accumuloConf = serverConf;
timeoutSupplier = memoizeWithExpiration(
() -> getConfiguration().getTimeInMillis(Property.GENERAL_RPC_TIMEOUT), 100, MILLISECONDS);
sslSupplier = Suppliers.memoize(() -> SslConnectionParams.forClient(getConfiguration()));
sslSupplier = memoize(() -> SslConnectionParams.forClient(getConfiguration()));
saslSupplier = memoizeWithExpiration(
() -> SaslConnectionParams.from(getConfiguration(), getCredentials().getToken()), 100,
MILLISECONDS);
Expand Down Expand Up @@ -333,7 +338,7 @@ public synchronized void setCredentials(Credentials newCredentials) {
*/
public AccumuloConfiguration getConfiguration() {
ensureOpen();
return serverConf;
return accumuloConf;
}

/**
Expand Down Expand Up @@ -484,26 +489,7 @@ public synchronized TCredentials rpcCreds() {
* @return a UUID
*/
public InstanceId getInstanceID() {
if (instanceId == null) {
// lookup by name
final String instanceName = info.getInstanceName();
String instanceNamePath = Constants.ZROOT + Constants.ZINSTANCES + "/" + instanceName;
byte[] data = zooCache.get(instanceNamePath);
if (data == null) {
throw new RuntimeException(
"Instance name " + instanceName + " does not exist in zookeeper. "
+ "Run \"accumulo org.apache.accumulo.server.util.ListInstances\" to see a list.");
}
String instanceIdString = new String(data, UTF_8);
// verify that the instanceId found via the instanceName actually exists as an instance
if (zooCache.get(Constants.ZROOT + "/" + instanceIdString) == null) {
throw new RuntimeException("Instance id " + instanceIdString
+ (instanceName == null ? "" : " pointed to by the name " + instanceName)
+ " does not exist in zookeeper");
}
instanceId = InstanceId.of(instanceIdString);
}
return instanceId;
return info.getInstanceId();
}

public String getZooKeeperRoot() {
Expand Down Expand Up @@ -541,7 +527,7 @@ public int getZooKeepersSessionTimeOut() {
}

public ZooCache getZooCache() {
return zooCache;
return zooCache.get();
}

private TableZooHelper tableZooHelper;
Expand Down Expand Up @@ -796,6 +782,9 @@ public AuthenticationToken token() {
@Override
public synchronized void close() {
closed = true;
if (zooKeeperOpened.get()) {
zooSession.get().close();
}
if (thriftTransportPool != null) {
thriftTransportPool.shutdown();
}
Expand All @@ -816,7 +805,7 @@ public static class ClientBuilderImpl<T>
SslOptions<T>, SaslOptions<T>, ClientFactory<T>, FromOptions<T> {

private Properties properties = new Properties();
private AuthenticationToken token = null;
private Optional<AuthenticationToken> tokenOpt = Optional.empty();
private final Function<ClientBuilderImpl<T>,T> builderFunction;
private UncaughtExceptionHandler ueh = null;

Expand All @@ -825,12 +814,9 @@ public ClientBuilderImpl(Function<ClientBuilderImpl<T>,T> builderFunction) {
}

private ClientInfo getClientInfo() {
if (token != null) {
ClientProperty.validate(properties, false);
return new ClientInfoImpl(properties, token);
}
ClientProperty.validate(properties);
return new ClientInfoImpl(properties);
// validate the token in the properties if not provided here
ClientProperty.validate(properties, tokenOpt.isEmpty());
return new ClientInfoImpl(properties, tokenOpt);
}

private UncaughtExceptionHandler getUncaughtExceptionHandler() {
Expand Down Expand Up @@ -1001,7 +987,7 @@ public ConnectionOptions<T> as(CharSequence principal, AuthenticationToken token
}
setProperty(ClientProperty.AUTH_PRINCIPAL, principal.toString());
ClientProperty.setAuthenticationToken(properties, token);
this.token = token;
this.tokenOpt = Optional.of(token);
return this;
}

Expand All @@ -1025,9 +1011,9 @@ public ClientFactory<T> withUncaughtExceptionHandler(UncaughtExceptionHandler ue

}

public ZooReader getZooReader() {
public ZooSession getZooSession() {
ensureOpen();
return zooReader;
return zooSession.get();
}

protected long getTransportPoolMaxAgeMillis() {
Expand Down Expand Up @@ -1069,7 +1055,14 @@ && getConfiguration().getBoolean(Property.GENERAL_MICROMETER_CACHE_METRICS_ENABL
public synchronized ZookeeperLockChecker getTServerLockChecker() {
ensureOpen();
if (this.zkLockChecker == null) {
this.zkLockChecker = new ZookeeperLockChecker(this);
// make this use its own ZooSession and ZooCache, because this is used by the
// tablet location cache, which is a static singleton reused by multiple clients
// so, it can't rely on being able to continue to use the same client's ZooCache,
// because that client could be closed, and its ZooSession also closed
// this needs to be fixed; TODO https://github.com/apache/accumulo/issues/2301
var zk = info.getZooKeeperSupplier(ZookeeperLockChecker.class.getSimpleName()).get();
this.zkLockChecker =
new ZookeeperLockChecker(new ZooCache(zk), getZooKeeperRoot(), getServerPaths());
}
return this.zkLockChecker;
}
Expand Down
Loading

0 comments on commit 21bd629

Please sign in to comment.