haridsv commented on code in PR #1691:
URL: https://github.com/apache/phoenix/pull/1691#discussion_r1344390663
##########
phoenix-core/src/main/protobuf/RegionServerEndpointService.proto:
##########
@@ -36,7 +36,24 @@ message LastDDLTimestampRequest {
required int64 lastDDLTimestamp = 4;
}
-service RegionServerCoprocService {
+message InvalidateServerMetadataCache {
+ // Will be HConstants.EMPTY_BYTE_ARRAY if tenantID or schema name is null.
+ required bytes tenantId = 1;
+ required bytes schemaName = 2;
+ required bytes tableName = 3;
Review Comment:
Since 3 fields are common with that of `LastDDLTimestampRequest`, we could
avoid duplication if we use composition (unfortunately, protobuf has no
inheritance).
##########
phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java:
##########
@@ -432,4 +434,17 @@ public static boolean isHBaseNamespaceAvailable(Admin
admin, String schemaName)
String[] hbaseNamespaces = admin.listNamespaces();
return Arrays.asList(hbaseNamespaces).contains(schemaName);
}
+
+ /**
+ * Convert ServiceException into an IOException
+ * @param se ServiceException
+ * @return IOException
+ */
+ public static IOException parseServiceException(ServiceException se) {
+ Throwable cause = se.getCause();
+ if (cause != null && cause instanceof IOException) {
+ return (IOException) cause;
+ }
+ return new IOException(se);
Review Comment:
Should this be `new IOException(cause)`?
##########
phoenix-core/src/main/java/org/apache/phoenix/coprocessor/PhoenixRegionServerEndpoint.java:
##########
@@ -73,4 +76,27 @@ public void validateLastDDLTimestamp(RpcController
controller,
}
}
}
+
+ @Override
+ public void invalidateServerMetadataCache(RpcController controller,
+ RegionServerEndpointProtos.InvalidateServerMetadataCacheRequest
request,
+
RpcCallback<RegionServerEndpointProtos.InvalidateServerMetadataCacheResponse>
done) {
+ for (RegionServerEndpointProtos.InvalidateServerMetadataCache
invalidateCacheRequest
+ : request.getInvalidateServerMetadataCacheRequestsList()) {
+ byte[] tenantID =
invalidateCacheRequest.getTenantId().toByteArray();
+ byte[] schemaName =
invalidateCacheRequest.getSchemaName().toByteArray();
+ byte[] tableName =
invalidateCacheRequest.getTableName().toByteArray();
+ String fullTableName = SchemaUtil.getTableName(schemaName,
tableName);
+ String tenantIDStr = Bytes.toString(tenantID);
Review Comment:
If there is composition at the protobuf level for the first 3 fields, then
these 5 lines that are common between this function and
`validateLastDDLTimestamp` could be processed as a common utility.
##########
phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java:
##########
@@ -3416,6 +3437,167 @@ private MetaDataMutationResult mutateColumn(
}
}
+ /**
+ * Invalidate metadata cache from all region servers for the given tenant
and table name.
+ * @param tenantId
+ * @param schemaName
+ * @param tableOrViewName
+ * @throws Throwable
+ */
+ private void invalidateServerMetadataCache(byte[] tenantId,
byte[]schemaName,
+ byte[] tableOrViewName) throws Throwable {
+ Configuration conf = env.getConfiguration();
+ String value = conf.get(REGIONSERVER_COPROCESSOR_CONF_KEY);
+ if (value == null
+ ||
!value.contains(PhoenixRegionServerEndpoint.class.getName())) {
+ // PhoenixRegionServerEndpoint is not loaded. We don't have to
invalidate the cache.
+ LOGGER.info("Skip invalidating server metadata cache for tenantID:
{},"
+ + " schema name: {}, table Name: {} since
PhoenixRegionServerEndpoint"
+ + " is not loaded", Bytes.toString(tenantId),
+ Bytes.toString(schemaName),
Bytes.toString(tableOrViewName));
+ return;
+ }
+ Properties properties = new Properties();
+ // Skip checking of system table existence since the system tables
should have created
+ // by now.
+ properties.setProperty(SKIP_SYSTEM_TABLES_EXISTENCE_CHECK, "true");
+ try (PhoenixConnection connection =
QueryUtil.getConnectionOnServer(properties,
+ env.getConfiguration()).unwrap(PhoenixConnection.class);
+ Admin admin = connection.getQueryServices().getAdmin()) {
+ // This will incur an extra RPC to the master. This RPC is
required since we want to
+ // get current list of regionservers.
+ Collection<ServerName> serverNames = admin.getRegionServers(true);
+ invalidateServerMetadataCacheWithRetries(admin, serverNames,
tenantId, schemaName,
+ tableOrViewName, false);
+ }
+ }
+
+ /**
+ * Invalidate metadata cache on all regionservers with retries for the
given tenantID
+ * and tableName with retries. We retry once before failing the operation.
+ *
+ * @param admin
+ * @param serverNames
+ * @param tenantId
+ * @param schemaName
+ * @param tableOrViewName
+ * @param isRetry
+ * @throws Throwable
+ */
+ private void invalidateServerMetadataCacheWithRetries(Admin admin,
+ Collection<ServerName> serverNames, byte[] tenantId, byte[]
schemaName,
+ byte[] tableOrViewName, boolean isRetry) throws Throwable {
+ String fullTableName = SchemaUtil.getTableName(schemaName,
tableOrViewName);
+ String tenantIDStr = Bytes.toString(tenantId);
+ LOGGER.info("Invalidating metadata cache for tenantID: {}, tableName:
{} for"
+ + " region servers: {}, isRetry: {}", tenantIDStr,
fullTableName,
+ serverNames, isRetry);
+ RegionServerEndpointProtos.InvalidateServerMetadataCacheRequest
request =
+ getRequest(tenantId, schemaName, tableOrViewName);
+ // TODO Do I need my own executor or can I re-use
QueryServices#Executor
+ // since it is supposed to be used only for scans according to
documentation?
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ Map<Future, ServerName> map = new HashMap<>();
+ for (ServerName serverName : serverNames) {
+ CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+ try {
+ PhoenixStopWatch innerWatch = new
PhoenixStopWatch().start();
+ // TODO Using the same as ServerCacheClient but need to
think if we need some
+ // special controller for invalidating cache since this is
in the path of
+ // DDL operations. We also need to think of we need
separate RPC handler
+ // threads for this?
+ ServerRpcController controller = new ServerRpcController();
+
RegionServerEndpointProtos.RegionServerEndpointService.BlockingInterface
+ service =
RegionServerEndpointProtos.RegionServerEndpointService
+
.newBlockingStub(admin.coprocessorService(serverName));
+ LOGGER.info("Sending invalidate metadata cache for
tenantID: {}, tableName: {}"
+ + " to region server: {}", tenantIDStr,
fullTableName, serverName);
+ // The timeout for this particular request is managed by
config parameter:
+ // hbase.rpc.timeout. Even if the future times out, this
runnable can be in
+ // RUNNING state and will not be interrupted.
+ service.invalidateServerMetadataCache(controller, request);
+ LOGGER.info("Invalidating metadata cache for tenantID: {},
tableName: {}"
+ + " on region server: {} completed successfully
and it took {} ms",
+ tenantIDStr, fullTableName, serverName,
+ innerWatch.stop().elapsedMillis());
+ // TODO Create a histogram metric for time taken for
invalidating the cache.
+ } catch (ServiceException se) {
+ LOGGER.error("Invalidating metadata cache for tenantID:
{}, tableName: {}"
+ + " failed for regionserver {}",
tenantIDStr, fullTableName,
+ serverName, se);
+ IOException ioe = ServerUtil.parseServiceException(se);
+ throw new CompletionException(ioe);
+ }
+ });
+ futures.add(future);
+ map.put(future, serverName);
+ }
+
+ // Here we create one master like future which tracks individual future
+ // for each region server.
+ CompletableFuture<Void> allFutures = CompletableFuture.allOf(
+ futures.toArray(new CompletableFuture[0]));
+ try {
+ allFutures.get(metadataCacheInvalidationTimeoutMs,
TimeUnit.MILLISECONDS);
+ } catch (Throwable t) {
+ List<ServerName> failedServers = getFailedServers(futures, map);
+ LOGGER.error("Invalidating metadata cache for tenantID: {},
tableName: {} failed for "
+ + "region servers: {}", tenantIDStr, fullTableName,
failedServers, t);
+ if (isRetry) {
+ // If this is a retry attempt then just fail the operation.
+ if (allFutures.isCompletedExceptionally()) {
+ if (t instanceof ExecutionException) {
+ t = t.getCause();
+ }
+ }
+ throw t;
+ } else {
+ // This is the first attempt, we can retry once.
+ // Indicate that this is a retry attempt.
+ invalidateServerMetadataCacheWithRetries(admin, failedServers,
+ tenantId, schemaName, tableOrViewName, true);
+ }
+ }
+ }
+
+ /*
+ Get the list of regionservers that failed the invalidateCache rpc.
+ */
+ private List<ServerName> getFailedServers(List<CompletableFuture<Void>>
futures,
+ Map<Future, ServerName> map) {
+ List<ServerName> failedServers = new ArrayList<>();
+ for (CompletableFuture completedFuture : futures) {
+ if (completedFuture.isDone() == false) {
Review Comment:
I am not familiar with futures, but what happens in case of a timeout? Will
it have a `false` value? In the timeout case, won't the remote RPC thread be in
a stuck state? What is the chance of a retry making progress, besides
initiating a concurrent request?
##########
phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java:
##########
@@ -3416,6 +3437,167 @@ private MetaDataMutationResult mutateColumn(
}
}
+ /**
+ * Invalidate metadata cache from all region servers for the given tenant
and table name.
+ * @param tenantId
+ * @param schemaName
+ * @param tableOrViewName
+ * @throws Throwable
+ */
+ private void invalidateServerMetadataCache(byte[] tenantId,
byte[]schemaName,
+ byte[] tableOrViewName) throws Throwable {
+ Configuration conf = env.getConfiguration();
+ String value = conf.get(REGIONSERVER_COPROCESSOR_CONF_KEY);
+ if (value == null
+ ||
!value.contains(PhoenixRegionServerEndpoint.class.getName())) {
+ // PhoenixRegionServerEndpoint is not loaded. We don't have to
invalidate the cache.
+ LOGGER.info("Skip invalidating server metadata cache for tenantID:
{},"
+ + " schema name: {}, table Name: {} since
PhoenixRegionServerEndpoint"
+ + " is not loaded", Bytes.toString(tenantId),
+ Bytes.toString(schemaName),
Bytes.toString(tableOrViewName));
+ return;
+ }
+ Properties properties = new Properties();
+ // Skip checking of system table existence since the system tables
should have created
+ // by now.
+ properties.setProperty(SKIP_SYSTEM_TABLES_EXISTENCE_CHECK, "true");
+ try (PhoenixConnection connection =
QueryUtil.getConnectionOnServer(properties,
+ env.getConfiguration()).unwrap(PhoenixConnection.class);
+ Admin admin = connection.getQueryServices().getAdmin()) {
+ // This will incur an extra RPC to the master. This RPC is
required since we want to
+ // get current list of regionservers.
+ Collection<ServerName> serverNames = admin.getRegionServers(true);
+ invalidateServerMetadataCacheWithRetries(admin, serverNames,
tenantId, schemaName,
+ tableOrViewName, false);
+ }
+ }
+
+ /**
+ * Invalidate metadata cache on all regionservers with retries for the
given tenantID
+ * and tableName with retries. We retry once before failing the operation.
+ *
+ * @param admin
+ * @param serverNames
+ * @param tenantId
+ * @param schemaName
+ * @param tableOrViewName
+ * @param isRetry
+ * @throws Throwable
+ */
+ private void invalidateServerMetadataCacheWithRetries(Admin admin,
+ Collection<ServerName> serverNames, byte[] tenantId, byte[]
schemaName,
+ byte[] tableOrViewName, boolean isRetry) throws Throwable {
+ String fullTableName = SchemaUtil.getTableName(schemaName,
tableOrViewName);
+ String tenantIDStr = Bytes.toString(tenantId);
+ LOGGER.info("Invalidating metadata cache for tenantID: {}, tableName:
{} for"
+ + " region servers: {}, isRetry: {}", tenantIDStr,
fullTableName,
+ serverNames, isRetry);
+ RegionServerEndpointProtos.InvalidateServerMetadataCacheRequest
request =
+ getRequest(tenantId, schemaName, tableOrViewName);
+ // TODO Do I need my own executor or can I re-use
QueryServices#Executor
+ // since it is supposed to be used only for scans according to
documentation?
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ Map<Future, ServerName> map = new HashMap<>();
+ for (ServerName serverName : serverNames) {
+ CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+ try {
+ PhoenixStopWatch innerWatch = new
PhoenixStopWatch().start();
+ // TODO Using the same as ServerCacheClient but need to
think if we need some
+ // special controller for invalidating cache since this is
in the path of
+ // DDL operations. We also need to think of we need
separate RPC handler
+ // threads for this?
+ ServerRpcController controller = new ServerRpcController();
+
RegionServerEndpointProtos.RegionServerEndpointService.BlockingInterface
+ service =
RegionServerEndpointProtos.RegionServerEndpointService
+
.newBlockingStub(admin.coprocessorService(serverName));
+ LOGGER.info("Sending invalidate metadata cache for
tenantID: {}, tableName: {}"
+ + " to region server: {}", tenantIDStr,
fullTableName, serverName);
+ // The timeout for this particular request is managed by
config parameter:
+ // hbase.rpc.timeout. Even if the future times out, this
runnable can be in
+ // RUNNING state and will not be interrupted.
+ service.invalidateServerMetadataCache(controller, request);
+ LOGGER.info("Invalidating metadata cache for tenantID: {},
tableName: {}"
+ + " on region server: {} completed successfully
and it took {} ms",
+ tenantIDStr, fullTableName, serverName,
+ innerWatch.stop().elapsedMillis());
+ // TODO Create a histogram metric for time taken for
invalidating the cache.
+ } catch (ServiceException se) {
+ LOGGER.error("Invalidating metadata cache for tenantID:
{}, tableName: {}"
+ + " failed for regionserver {}",
tenantIDStr, fullTableName,
+ serverName, se);
+ IOException ioe = ServerUtil.parseServiceException(se);
+ throw new CompletionException(ioe);
+ }
+ });
+ futures.add(future);
+ map.put(future, serverName);
+ }
+
+ // Here we create one master like future which tracks individual future
+ // for each region server.
+ CompletableFuture<Void> allFutures = CompletableFuture.allOf(
+ futures.toArray(new CompletableFuture[0]));
+ try {
+ allFutures.get(metadataCacheInvalidationTimeoutMs,
TimeUnit.MILLISECONDS);
Review Comment:
Perhaps we should have a discount for retries?
##########
phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java:
##########
@@ -3416,6 +3437,167 @@ private MetaDataMutationResult mutateColumn(
}
}
+ /**
+ * Invalidate metadata cache from all region servers for the given tenant
and table name.
+ * @param tenantId
+ * @param schemaName
+ * @param tableOrViewName
+ * @throws Throwable
+ */
+ private void invalidateServerMetadataCache(byte[] tenantId,
byte[]schemaName,
+ byte[] tableOrViewName) throws Throwable {
+ Configuration conf = env.getConfiguration();
+ String value = conf.get(REGIONSERVER_COPROCESSOR_CONF_KEY);
+ if (value == null
+ ||
!value.contains(PhoenixRegionServerEndpoint.class.getName())) {
+ // PhoenixRegionServerEndpoint is not loaded. We don't have to
invalidate the cache.
+ LOGGER.info("Skip invalidating server metadata cache for tenantID:
{},"
+ + " schema name: {}, table Name: {} since
PhoenixRegionServerEndpoint"
+ + " is not loaded", Bytes.toString(tenantId),
+ Bytes.toString(schemaName),
Bytes.toString(tableOrViewName));
+ return;
+ }
+ Properties properties = new Properties();
+ // Skip checking of system table existence since the system tables
should have created
+ // by now.
+ properties.setProperty(SKIP_SYSTEM_TABLES_EXISTENCE_CHECK, "true");
+ try (PhoenixConnection connection =
QueryUtil.getConnectionOnServer(properties,
+ env.getConfiguration()).unwrap(PhoenixConnection.class);
+ Admin admin = connection.getQueryServices().getAdmin()) {
+ // This will incur an extra RPC to the master. This RPC is
required since we want to
+ // get current list of regionservers.
+ Collection<ServerName> serverNames = admin.getRegionServers(true);
+ invalidateServerMetadataCacheWithRetries(admin, serverNames,
tenantId, schemaName,
+ tableOrViewName, false);
+ }
+ }
+
+ /**
+ * Invalidate metadata cache on all regionservers with retries for the
given tenantID
+ * and tableName with retries. We retry once before failing the operation.
+ *
+ * @param admin
+ * @param serverNames
+ * @param tenantId
+ * @param schemaName
+ * @param tableOrViewName
+ * @param isRetry
+ * @throws Throwable
+ */
+ private void invalidateServerMetadataCacheWithRetries(Admin admin,
+ Collection<ServerName> serverNames, byte[] tenantId, byte[]
schemaName,
+ byte[] tableOrViewName, boolean isRetry) throws Throwable {
+ String fullTableName = SchemaUtil.getTableName(schemaName,
tableOrViewName);
+ String tenantIDStr = Bytes.toString(tenantId);
+ LOGGER.info("Invalidating metadata cache for tenantID: {}, tableName:
{} for"
+ + " region servers: {}, isRetry: {}", tenantIDStr,
fullTableName,
+ serverNames, isRetry);
+ RegionServerEndpointProtos.InvalidateServerMetadataCacheRequest
request =
+ getRequest(tenantId, schemaName, tableOrViewName);
+ // TODO Do I need my own executor or can I re-use
QueryServices#Executor
+ // since it is supposed to be used only for scans according to
documentation?
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ Map<Future, ServerName> map = new HashMap<>();
+ for (ServerName serverName : serverNames) {
+ CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+ try {
+ PhoenixStopWatch innerWatch = new
PhoenixStopWatch().start();
+ // TODO Using the same as ServerCacheClient but need to
think if we need some
+ // special controller for invalidating cache since this is
in the path of
+ // DDL operations. We also need to think of we need
separate RPC handler
+ // threads for this?
+ ServerRpcController controller = new ServerRpcController();
+
RegionServerEndpointProtos.RegionServerEndpointService.BlockingInterface
+ service =
RegionServerEndpointProtos.RegionServerEndpointService
+
.newBlockingStub(admin.coprocessorService(serverName));
+ LOGGER.info("Sending invalidate metadata cache for
tenantID: {}, tableName: {}"
+ + " to region server: {}", tenantIDStr,
fullTableName, serverName);
+ // The timeout for this particular request is managed by
config parameter:
+ // hbase.rpc.timeout. Even if the future times out, this
runnable can be in
+ // RUNNING state and will not be interrupted.
+ service.invalidateServerMetadataCache(controller, request);
+ LOGGER.info("Invalidating metadata cache for tenantID: {},
tableName: {}"
+ + " on region server: {} completed successfully
and it took {} ms",
+ tenantIDStr, fullTableName, serverName,
+ innerWatch.stop().elapsedMillis());
+ // TODO Create a histogram metric for time taken for
invalidating the cache.
+ } catch (ServiceException se) {
+ LOGGER.error("Invalidating metadata cache for tenantID:
{}, tableName: {}"
+ + " failed for regionserver {}",
tenantIDStr, fullTableName,
+ serverName, se);
+ IOException ioe = ServerUtil.parseServiceException(se);
+ throw new CompletionException(ioe);
+ }
+ });
+ futures.add(future);
+ map.put(future, serverName);
+ }
+
+ // Here we create one master like future which tracks individual future
+ // for each region server.
+ CompletableFuture<Void> allFutures = CompletableFuture.allOf(
+ futures.toArray(new CompletableFuture[0]));
+ try {
+ allFutures.get(metadataCacheInvalidationTimeoutMs,
TimeUnit.MILLISECONDS);
+ } catch (Throwable t) {
+ List<ServerName> failedServers = getFailedServers(futures, map);
+ LOGGER.error("Invalidating metadata cache for tenantID: {},
tableName: {} failed for "
Review Comment:
This should be an error only if we are not going to retry?
##########
phoenix-core/src/main/protobuf/RegionServerEndpointService.proto:
##########
@@ -36,7 +36,24 @@ message LastDDLTimestampRequest {
required int64 lastDDLTimestamp = 4;
}
-service RegionServerCoprocService {
+message InvalidateServerMetadataCache {
Review Comment:
Just curious, how is the backwards compatibility handled for these sort of
changes?
##########
phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java:
##########
@@ -3416,6 +3437,167 @@ private MetaDataMutationResult mutateColumn(
}
}
+ /**
+ * Invalidate metadata cache from all region servers for the given tenant
and table name.
+ * @param tenantId
+ * @param schemaName
+ * @param tableOrViewName
+ * @throws Throwable
+ */
+ private void invalidateServerMetadataCache(byte[] tenantId,
byte[]schemaName,
+ byte[] tableOrViewName) throws Throwable {
+ Configuration conf = env.getConfiguration();
+ String value = conf.get(REGIONSERVER_COPROCESSOR_CONF_KEY);
+ if (value == null
+ ||
!value.contains(PhoenixRegionServerEndpoint.class.getName())) {
+ // PhoenixRegionServerEndpoint is not loaded. We don't have to
invalidate the cache.
+ LOGGER.info("Skip invalidating server metadata cache for tenantID:
{},"
+ + " schema name: {}, table Name: {} since
PhoenixRegionServerEndpoint"
+ + " is not loaded", Bytes.toString(tenantId),
+ Bytes.toString(schemaName),
Bytes.toString(tableOrViewName));
+ return;
+ }
+ Properties properties = new Properties();
+ // Skip checking of system table existence since the system tables
should have created
+ // by now.
+ properties.setProperty(SKIP_SYSTEM_TABLES_EXISTENCE_CHECK, "true");
+ try (PhoenixConnection connection =
QueryUtil.getConnectionOnServer(properties,
+ env.getConfiguration()).unwrap(PhoenixConnection.class);
+ Admin admin = connection.getQueryServices().getAdmin()) {
+ // This will incur an extra RPC to the master. This RPC is
required since we want to
+ // get current list of regionservers.
+ Collection<ServerName> serverNames = admin.getRegionServers(true);
+ invalidateServerMetadataCacheWithRetries(admin, serverNames,
tenantId, schemaName,
+ tableOrViewName, false);
+ }
+ }
+
+ /**
+ * Invalidate metadata cache on all regionservers with retries for the
given tenantID
+ * and tableName with retries. We retry once before failing the operation.
+ *
+ * @param admin
+ * @param serverNames
+ * @param tenantId
+ * @param schemaName
+ * @param tableOrViewName
+ * @param isRetry
+ * @throws Throwable
+ */
+ private void invalidateServerMetadataCacheWithRetries(Admin admin,
+ Collection<ServerName> serverNames, byte[] tenantId, byte[]
schemaName,
+ byte[] tableOrViewName, boolean isRetry) throws Throwable {
+ String fullTableName = SchemaUtil.getTableName(schemaName,
tableOrViewName);
+ String tenantIDStr = Bytes.toString(tenantId);
+ LOGGER.info("Invalidating metadata cache for tenantID: {}, tableName:
{} for"
+ + " region servers: {}, isRetry: {}", tenantIDStr,
fullTableName,
+ serverNames, isRetry);
+ RegionServerEndpointProtos.InvalidateServerMetadataCacheRequest
request =
+ getRequest(tenantId, schemaName, tableOrViewName);
+ // TODO Do I need my own executor or can I re-use
QueryServices#Executor
+ // since it is supposed to be used only for scans according to
documentation?
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ Map<Future, ServerName> map = new HashMap<>();
+ for (ServerName serverName : serverNames) {
+ CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+ try {
+ PhoenixStopWatch innerWatch = new
PhoenixStopWatch().start();
+ // TODO Using the same as ServerCacheClient but need to
think if we need some
+ // special controller for invalidating cache since this is
in the path of
+ // DDL operations. We also need to think of we need
separate RPC handler
+ // threads for this?
+ ServerRpcController controller = new ServerRpcController();
+
RegionServerEndpointProtos.RegionServerEndpointService.BlockingInterface
+ service =
RegionServerEndpointProtos.RegionServerEndpointService
+
.newBlockingStub(admin.coprocessorService(serverName));
+ LOGGER.info("Sending invalidate metadata cache for
tenantID: {}, tableName: {}"
+ + " to region server: {}", tenantIDStr,
fullTableName, serverName);
+ // The timeout for this particular request is managed by
config parameter:
+ // hbase.rpc.timeout. Even if the future times out, this
runnable can be in
+ // RUNNING state and will not be interrupted.
+ service.invalidateServerMetadataCache(controller, request);
+ LOGGER.info("Invalidating metadata cache for tenantID: {},
tableName: {}"
+ + " on region server: {} completed successfully
and it took {} ms",
+ tenantIDStr, fullTableName, serverName,
+ innerWatch.stop().elapsedMillis());
+ // TODO Create a histogram metric for time taken for
invalidating the cache.
+ } catch (ServiceException se) {
+ LOGGER.error("Invalidating metadata cache for tenantID:
{}, tableName: {}"
+ + " failed for regionserver {}",
tenantIDStr, fullTableName,
+ serverName, se);
+ IOException ioe = ServerUtil.parseServiceException(se);
+ throw new CompletionException(ioe);
+ }
+ });
+ futures.add(future);
+ map.put(future, serverName);
+ }
+
+ // Here we create one master like future which tracks individual future
+ // for each region server.
+ CompletableFuture<Void> allFutures = CompletableFuture.allOf(
+ futures.toArray(new CompletableFuture[0]));
+ try {
+ allFutures.get(metadataCacheInvalidationTimeoutMs,
TimeUnit.MILLISECONDS);
+ } catch (Throwable t) {
Review Comment:
This catches every kind of exception, is it desirable to retry in error
cases? Why not be specific to those that are just declared?
```suggestion
} catch (CancellationException | ExecutionException |
InterruptedException | TimeoutException t) {
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
