keith-turner commented on code in PR #5875:
URL: https://github.com/apache/accumulo/pull/5875#discussion_r2344450555
##########
core/src/main/java/org/apache/accumulo/core/client/admin/servers/ServerId.java:
##########
@@ -18,42 +18,167 @@
*/
package org.apache.accumulo.core.client.admin.servers;
+import static org.apache.accumulo.core.util.LazySingletons.GSON;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.Serializable;
import java.util.Objects;
import org.apache.accumulo.core.conf.PropertyType.PortRange;
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;
+import com.google.common.net.HostAndPort;
/**
* Object representing the type, resource group, and address of a server
process.
*
* @since 4.0.0
*/
-public final class ServerId implements Comparable<ServerId> {
+public final class ServerId implements Comparable<ServerId>, Serializable {
+
+ private static final long serialVersionUID = 1L;
/**
* Server process type names.
*
* @since 4.0.0
*/
public enum Type {
- MANAGER, MONITOR, GARBAGE_COLLECTOR, COMPACTOR, SCAN_SERVER, TABLET_SERVER;
+ MANAGER, MINI, MONITOR, GARBAGE_COLLECTOR, COMPACTOR, SCAN_SERVER,
TABLET_SERVER;
+ }
+
+ public static record ServerIdInfo(String type, String resourceGroup, String
host, int port) {
+ public ServerId getServerId() {
+ return new 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();
+
+ private static ServerId resolve(ServerIdInfo info) {
+ return cache.get(info, k -> info.getServerId());
+ }
+
+ public static ServerId compactor(HostAndPort hp) {
+ return resolve(new ServerIdInfo(Type.COMPACTOR.name(),
ResourceGroupId.DEFAULT.canonical(),
+ hp.getHost(), hp.getPort()));
+ }
+
+ 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 compactor(ResourceGroupId rgid, HostAndPort hp) {
+ return resolve(
+ new ServerIdInfo(Type.COMPACTOR.name(), rgid.canonical(),
hp.getHost(), hp.getPort()));
+ }
+
+ 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(HostAndPort hp) {
+ return resolve(new ServerIdInfo(Type.SCAN_SERVER.name(),
ResourceGroupId.DEFAULT.canonical(),
+ hp.getHost(), hp.getPort()));
+ }
+
+ 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 sserver(ResourceGroupId rgid, HostAndPort hp) {
+ return resolve(
+ new ServerIdInfo(Type.SCAN_SERVER.name(), rgid.canonical(),
hp.getHost(), hp.getPort()));
+ }
+
+ public static ServerId tserver(HostAndPort hp) {
+ return resolve(new ServerIdInfo(Type.TABLET_SERVER.name(),
ResourceGroupId.DEFAULT.canonical(),
+ hp.getHost(), hp.getPort()));
+ }
+
+ 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 tserver(ResourceGroupId rgid, HostAndPort hp) {
+ return resolve(
+ new ServerIdInfo(Type.TABLET_SERVER.name(), rgid.canonical(),
hp.getHost(), hp.getPort()));
+ }
+
+ public static ServerId dynamic(Type type, ResourceGroupId rgid, HostAndPort
hp) {
+ return resolve(new ServerIdInfo(type.name(), rgid.canonical(),
hp.getHost(), hp.getPort()));
+ }
+
+ 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) {
+ String parts[] = name.split("\\+");
+ Preconditions.checkArgument(parts.length == 2, "Invalid server id in wal
file: " + name);
+ // return an uncached tserver object
+ return ServerId.tserver(parts[0], Integer.parseInt(parts[1]));
+ }
+
+ public static final ServerId deserialize(String json) {
+ return GSON.get().fromJson(json, ServerIdInfo.class).getServerId();
}
private final Type type;
private final ResourceGroupId resourceGroup;
private final String host;
private final int port;
+ private transient HostAndPort hostPort;
- public ServerId(Type type, ResourceGroupId resourceGroup, String host, int
port) {
+ private ServerId(Type type, ResourceGroupId resourceGroup, String host, int
port) {
Review Comment:
If we could also get the session id added to this type then we would know
everything about a server. Not a change for this PR, but I suspect this PR
would make that easier to do. May be able to replace TServerInstance type w/
ServerId if that were done.
##########
core/src/main/java/org/apache/accumulo/core/client/admin/servers/ServerId.java:
##########
@@ -18,42 +18,167 @@
*/
package org.apache.accumulo.core.client.admin.servers;
+import static org.apache.accumulo.core.util.LazySingletons.GSON;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.Serializable;
import java.util.Objects;
import org.apache.accumulo.core.conf.PropertyType.PortRange;
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;
+import com.google.common.net.HostAndPort;
/**
* Object representing the type, resource group, and address of a server
process.
*
* @since 4.0.0
*/
-public final class ServerId implements Comparable<ServerId> {
+public final class ServerId implements Comparable<ServerId>, Serializable {
+
+ private static final long serialVersionUID = 1L;
/**
* Server process type names.
*
* @since 4.0.0
*/
public enum Type {
- MANAGER, MONITOR, GARBAGE_COLLECTOR, COMPACTOR, SCAN_SERVER, TABLET_SERVER;
+ MANAGER, MINI, MONITOR, GARBAGE_COLLECTOR, COMPACTOR, SCAN_SERVER,
TABLET_SERVER;
+ }
+
+ public static record ServerIdInfo(String type, String resourceGroup, String
host, int port) {
Review Comment:
ServerIdInfo seems like its needed for implementation of caching and
serialization, but maybe not intended to be in the public API? Could move it
outside f the public API or make it package private if that is the case.
No idea if this makes sense or would be workable, but looking at this wonder
if ServerId should be a record type itself w/ supporting static methods some
that are public API and some that are not.
##########
core/src/main/java/org/apache/accumulo/core/client/admin/servers/ServerId.java:
##########
@@ -72,6 +197,13 @@ public int getPort() {
return port;
}
+ public synchronized HostAndPort getHostPort() {
Review Comment:
May not want to introduce the HostAndPort type into the public API.
##########
core/src/main/java/org/apache/accumulo/core/client/TimedOutException.java:
##########
@@ -50,7 +52,7 @@ public TimedOutException(String msg) {
super(msg);
}
- public Set<String> getTimedOutSevers() {
+ public Set<ServerId> getTimedOutSevers() {
Review Comment:
Could leave this as string since this is a public API method
##########
core/src/main/java/org/apache/accumulo/core/client/MutationsRejectedException.java:
##########
@@ -108,7 +109,7 @@ public Map<TabletId,Set<SecurityErrorCode>>
getSecurityErrorCodes() {
* @return A list of servers that had internal errors when mutations were
written
*
*/
- public Collection<String> getErrorServers() {
+ public Collection<ServerId> getErrorServers() {
Review Comment:
This method is public API. Its in a dark corner of the API, could change it.
May be easiest to just leave itas string.
##########
server/manager/src/main/java/org/apache/accumulo/manager/TabletGroupWatcher.java:
##########
@@ -859,11 +859,11 @@ private void hostSuspendedTablet(TabletLists tLists,
TabletMetadata tm, Location
< tableConf.getTimeInMillis(Property.TABLE_SUSPEND_DURATION)) {
// Tablet is suspended. See if its tablet server is back.
TServerInstance returnInstance = null;
- Iterator<TServerInstance> find = tLists.destinations
- .tailMap(new TServerInstance(tm.getSuspend().server, "
")).keySet().iterator();
+ Iterator<TServerInstance> find =
+
tLists.destinations.tailMap(tm.getSuspend().server).keySet().iterator();
Review Comment:
Probably still need to create the TServerInstance w/ the empty session
string. This code is trying to find any server where the host and port match,
but the sesssion id does not matter. The empty session string sorts before all.
This code needs to be refactored so it can be unit tested in another PR.
Wonder if the existing code works as it depends heavily on how those sort,
which could have changed after this code was written.
```suggestion
tLists.destinations.tailMap(new
TServerInstance(tm.getSuspend().server, " ")).keySet().iterator();
```
##########
core/src/main/java/org/apache/accumulo/core/client/admin/servers/ServerId.java:
##########
@@ -116,7 +248,24 @@ public String toString() {
+ ", port= " + port + "]";
}
+ public String toWalFileName() {
Review Comment:
This seems like an implementation thing that should be moved out of the
public API, maybe to a static method somewhere.
##########
core/src/main/java/org/apache/accumulo/core/client/admin/Locations.java:
##########
@@ -50,5 +51,5 @@ public interface Locations {
*
* @return A tablet server location in the form of {@code <host>:<port>}
*/
- String getTabletLocation(TabletId tabletId);
+ ServerId getTabletLocation(TabletId tabletId);
Review Comment:
Could leave this as string also since its public API
##########
core/src/main/java/org/apache/accumulo/core/spi/balancer/data/TabletServerId.java:
##########
@@ -18,13 +18,14 @@
*/
package org.apache.accumulo.core.spi.balancer.data;
+import org.apache.accumulo.core.client.admin.servers.ServerId;
+
/**
* @since 2.1.0
*/
public interface TabletServerId extends Comparable<TabletServerId> {
- String getHost();
- int getPort();
Review Comment:
If we had the session in the ServerId, then this whole type in the spi could
go away.
##########
core/src/main/java/org/apache/accumulo/core/clientImpl/ConditionalWriterImpl.java:
##########
@@ -113,7 +113,7 @@ private static class ServerQueue {
boolean taskQueued = false;
}
- private final Map<String,ServerQueue> serverQueues;
+ private final Map<ServerId,ServerQueue> serverQueues;
Review Comment:
This is really nice, changing from string to ServerId. These changes can be
tricky though because some of the map methods take Object so it will still
compile when a string is passed. My IDE warns on these types of mismatches.
Wonder if anything in the maven build will do that.
--
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
