loserwang1024 commented on code in PR #1928:
URL: https://github.com/apache/fluss/pull/1928#discussion_r2489152420
##########
fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupSender.java:
##########
@@ -382,4 +395,47 @@ void initiateClose() {
lookupQueue.close();
running = false;
}
+
+ private void handleFetchLogExceptionForBucket(TableBucket tb, int
destination, ApiError error) {
+ ApiException exception = error.error().exception();
+ LOG.error("Failed to lookup from node {} for bucket {}", destination,
tb, exception);
+ if (exception instanceof InvalidMetadataException) {
+ LOG.warn(
+ "Invalid metadata error in fetch log request. "
+ + "Going to request metadata update.",
+ exception);
+ long tableId = tb.getTableId();
+ TableOrPartitions tableOrPartitions;
+ if (tb.getPartitionId() == null) {
+ tableOrPartitions = new
TableOrPartitions(Collections.singleton(tableId), null);
+ } else {
+ tableOrPartitions =
+ new TableOrPartitions(
+ null,
+ Collections.singleton(
+ new TablePartition(tableId,
tb.getPartitionId())));
+ }
+ invalidTableOrPartitions(tableOrPartitions);
+ }
+ }
+
+ /** A helper class to hold table ids or table partitions. */
+ private static class TableOrPartitions {
+ private final @Nullable Set<Long> tableIds;
+ private final @Nullable Set<TablePartition> tablePartitions;
+
+ TableOrPartitions(
+ @Nullable Set<Long> tableIds, @Nullable Set<TablePartition>
tablePartitions) {
+ this.tableIds = tableIds;
+ this.tablePartitions = tablePartitions;
+ }
+ }
+
Review Comment:
The code is same as LogFetcher, can we extract this same code in protected
class? If some problem occurs, we only need to change once.
--
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]
