dlmarion commented on code in PR #5875: URL: https://github.com/apache/accumulo/pull/5875#discussion_r2363192814
########## core/src/main/java/org/apache/accumulo/core/clientImpl/ServerIdUtil.java: ########## @@ -0,0 +1,125 @@ +/* + * 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.clientImpl; + +import org.apache.accumulo.core.client.admin.servers.ServerId; +import org.apache.accumulo.core.client.admin.servers.ServerId.Type; +import org.apache.accumulo.core.data.ResourceGroupId; +import org.apache.accumulo.core.util.cache.Caches; +import org.apache.accumulo.core.util.cache.Caches.CacheName; + +import com.github.benmanes.caffeine.cache.Cache; +import com.google.common.base.Preconditions; + +public class ServerIdUtil { + + public static record ServerIdInfo(String type, String resourceGroup, String host, int port) { + public ServerId getServerId() { + return new ServerId(ServerId.Type.valueOf(type), ResourceGroupId.of(resourceGroup), host, + port); + } + } + + // cache is for canonicalization/deduplication of created objects, + // to limit the number of ServerId objects in the JVM at any given moment + // WeakReferences are used because we don't need them to stick around any longer than they need to + private static final Cache<ServerIdInfo,ServerId> cache = + Caches.getInstance().createNewBuilder(CacheName.SERVER_ID, false).weakValues().build(); + + public static ServerIdInfo toServerIdInfo(ServerId server) { + return new ServerIdInfo(server.getType().name(), server.getResourceGroup().canonical(), + server.getHost(), server.getPort()); + } + + private static ServerId resolve(ServerIdInfo info) { + return cache.get(info, k -> info.getServerId()); + } + + public static ServerId compactor(String host, int port) { + return resolve( + new ServerIdInfo(Type.COMPACTOR.name(), ResourceGroupId.DEFAULT.canonical(), host, port)); + } + + public static ServerId compactor(ResourceGroupId rgid, String host, int port) { + return resolve(new ServerIdInfo(Type.COMPACTOR.name(), rgid.canonical(), host, port)); + } + + public static ServerId gc(String host, int port) { + return resolve(new ServerIdInfo(Type.GARBAGE_COLLECTOR.name(), + ResourceGroupId.DEFAULT.canonical(), host, port)); + } + + public static ServerId manager(String host, int port) { + return resolve( + new ServerIdInfo(Type.MANAGER.name(), ResourceGroupId.DEFAULT.canonical(), host, port)); + } + + public static ServerId mini(String host, int port) { + return resolve( + new ServerIdInfo(Type.MINI.name(), ResourceGroupId.DEFAULT.canonical(), host, port)); + } + + public static ServerId monitor(String host, int port) { + return resolve( + new ServerIdInfo(Type.MONITOR.name(), ResourceGroupId.DEFAULT.canonical(), host, port)); + } + + public static ServerId sserver(String host, int port) { + return resolve( + new ServerIdInfo(Type.SCAN_SERVER.name(), ResourceGroupId.DEFAULT.canonical(), host, port)); + } + + public static ServerId sserver(ResourceGroupId rgid, String host, int port) { + return resolve(new ServerIdInfo(Type.SCAN_SERVER.name(), rgid.canonical(), host, port)); + } + + public static ServerId tserver(String host, int port) { + return resolve(new ServerIdInfo(Type.TABLET_SERVER.name(), ResourceGroupId.DEFAULT.canonical(), + host, port)); + } + + public static ServerId tserver(ResourceGroupId rgid, String host, int port) { + return resolve(new ServerIdInfo(Type.TABLET_SERVER.name(), rgid.canonical(), host, port)); + } + + public static ServerId dynamic(Type type, ResourceGroupId rgid, String host, int port) { + return resolve(new ServerIdInfo(type.name(), rgid.canonical(), host, port)); + } + + public static ServerId fromWalFileName(String name) { + // Old path was host+port + // New path is group+host+port + // Need to handle both + String parts[] = name.split("\\+"); + Preconditions.checkArgument(parts.length == 2 || parts.length == 3, + "Invalid server id in wal file: " + name); + if (parts.length == 2) { + // return an uncached tserver object + return tserver(parts[0], Integer.parseInt(parts[1])); Review Comment: This is called from `LogEntry.validatedLogEntry` which returns a LogEntry object. The log path is stored in the tablet metadata in `LogColumnFamily`. This code here handles the current (not new) serialization of the server being just host+port. Creating the ServerId object here **assumes** the default ResourceGroup because we don't have any further information. This assumption could be dangerous if the LogEntry's ResourceGroup is used for comparison purposes in code that uses LogEntry. ########## core/src/main/java/org/apache/accumulo/core/metadata/TServerInstance.java: ########## @@ -32,54 +39,84 @@ * Therefore tablet assignments can be considered out-of-date if the tablet server instance * information has been changed. */ -public class TServerInstance implements Comparable<TServerInstance> { +public class TServerInstance implements Comparable<TServerInstance>, Serializable { - private final HostAndPort hostAndPort; - private final String hostPort; - private final String session; - private final String hostPortSession; + private static final long serialVersionUID = 1L; - public TServerInstance(HostAndPort address, String session) { - this.hostAndPort = address; - this.session = session; - this.hostPort = hostAndPort.toString(); - this.hostPortSession = hostPort + "[" + session + "]"; + public static record TServerInstanceInfo(ServerIdUtil.ServerIdInfo server, String session) { + public TServerInstance getTSI() { + return new TServerInstance(server.getServerId(), session); + } } - public TServerInstance(String formattedString) { - int pos = formattedString.indexOf("["); - if (pos < 0 || !formattedString.endsWith("]")) { - throw new IllegalArgumentException(formattedString); - } - this.hostAndPort = HostAndPort.fromString(formattedString.substring(0, pos)); - this.session = formattedString.substring(pos + 1, formattedString.length() - 1); - this.hostPort = hostAndPort.toString(); - this.hostPortSession = hostPort + "[" + session + "]"; + public static TServerInstance deserialize(String json) { + return GSON.get().fromJson(json, TServerInstanceInfo.class).getTSI(); + } + + public static TServerInstance fromZooKeeperPathString(String zkPath) { + + // TODO: WAL marker serializations using the old format could present a + // problem. If we change the code here to handle the old format, then + // we have to make a guess at the resource group, which could affect + // callers of this method (GcWalsFilter, WalStateManager, LiveTServerSet) Review Comment: This is another area of concern for dealing with serialized information. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: notifications-unsubscr...@accumulo.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
