Skip to content

[client][server] Preserve fetch bucket order to prevent starvation #4785

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fxbing wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[client][server] Preserve fetch bucket order to prevent starvation #4785

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
107 changes: 45 additions & 62 deletions fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -567,10 +567,7 @@ private void pendRemoteFetches(

@VisibleForTesting
Map<Integer, FetchLogRequest> prepareFetchLogRequests(List<TableBucket> fetchableBuckets) {
// Outer key: leaderId, inner key: tableId -> list of bucket requests
Map<Integer, Map<Long, List<PbFetchLogReqForBucket>>> fetchReqsByLeaderAndTable =
new HashMap<>();
int readyForFetchCount = 0;
Map<Integer, List<PbFetchLogReqForTable>> fetchReqsByLeader = new HashMap<>();
for (TableBucket tb : fetchableBuckets) {
Long offset = logScannerStatus.getBucketOffset(tb);
if (offset == null) {
Expand All @@ -595,60 +592,24 @@ Map<Integer, FetchLogRequest> prepareFetchLogRequests(List<TableBucket> fetchabl
tb,
leader);
} else {
PbFetchLogReqForBucket fetchLogReqForBucket =
new PbFetchLogReqForBucket()
.setBucketId(tb.getBucket())
.setFetchOffset(offset)
.setMaxFetchBytes(maxBucketFetchBytes);
if (tb.getPartitionId() != null) {
fetchLogReqForBucket.setPartitionId(tb.getPartitionId());
long tableId = tb.getTableId();
TableReadContext trc = tableReadContexts.get(tableId);
if (trc == null) {
LOG.warn(
"Skip fetch log request for table id {} because its read context "
+ "is not registered. The table may have already been "
+ "unregistered.",
tableId);
continue;
}
metadataUpdater
.getCluster()
.getBucketCount(TableOrPartition.of(tb.getTableId(), tb.getPartitionId()))
.ifPresent(fetchLogReqForBucket::setRoutingBucketCount);
fetchReqsByLeaderAndTable
.computeIfAbsent(leader, k -> new HashMap<>())
.computeIfAbsent(tb.getTableId(), k -> new ArrayList<>())
.add(fetchLogReqForBucket);
readyForFetchCount++;
}
}

if (readyForFetchCount == 0) {
return Collections.emptyMap();
} else {
Map<Integer, FetchLogRequest> fetchLogRequests = new HashMap<>();
for (Map.Entry<Integer, Map<Long, List<PbFetchLogReqForBucket>>> leaderEntry :
fetchReqsByLeaderAndTable.entrySet()) {
int leaderId = leaderEntry.getKey();
Map<Long, List<PbFetchLogReqForBucket>> tableReqs = leaderEntry.getValue();

FetchLogRequest fetchLogRequest =
new FetchLogRequest()
.setFollowerServerId(-1)
.setMaxBytes(maxFetchBytes)
.setMinBytes(minFetchBytes)
.setMaxWaitMs(maxFetchWaitMs)
.setReadPreference(readPreference.value());

List<PbFetchLogReqForTable> tablesReqs = new ArrayList<>();
for (Map.Entry<Long, List<PbFetchLogReqForBucket>> tableEntry :
tableReqs.entrySet()) {
long tableId = tableEntry.getKey();
List<PbFetchLogReqForBucket> bucketReqs = tableEntry.getValue();
TableReadContext trc = tableReadContexts.get(tableId);
if (trc == null) {
LOG.warn(
"Skip fetch log request for table id {} because its read context "
+ "is not registered. The table may have already been "
+ "unregistered.",
tableId);
continue;
}

PbFetchLogReqForTable reqForTable =
new PbFetchLogReqForTable().setTableId(tableId);
List<PbFetchLogReqForTable> tablesReqs =
fetchReqsByLeader.computeIfAbsent(leader, k -> new ArrayList<>());
PbFetchLogReqForTable reqForTable =
tablesReqs.isEmpty() ? null : tablesReqs.get(tablesReqs.size() - 1);
// Only group consecutive buckets of the same table to preserve the fetch rotation.
if (reqForTable == null || reqForTable.getTableId() != tableId) {
reqForTable = new PbFetchLogReqForTable().setTableId(tableId);
if (trc.readContext.isProjectionPushDowned()) {
assert trc.projection != null;
reqForTable
Expand All @@ -661,17 +622,39 @@ Map<Integer, FetchLogRequest> prepareFetchLogRequests(List<TableBucket> fetchabl
reqForTable.setFilterPredicate(trc.cachedPbPredicate);
reqForTable.setFilterSchemaId(trc.filterSchemaId);
}
reqForTable.addAllBucketsReqs(bucketReqs);
tablesReqs.add(reqForTable);
}
if (tablesReqs.isEmpty()) {
continue;

PbFetchLogReqForBucket fetchLogReqForBucket =
reqForTable
.addBucketsReq()
.setBucketId(tb.getBucket())
.setFetchOffset(offset)
.setMaxFetchBytes(maxBucketFetchBytes);
if (tb.getPartitionId() != null) {
fetchLogReqForBucket.setPartitionId(tb.getPartitionId());
}
fetchLogRequest.addAllTablesReqs(tablesReqs);
fetchLogRequests.put(leaderId, fetchLogRequest);
metadataUpdater
.getCluster()
.getBucketCount(TableOrPartition.of(tb.getTableId(), tb.getPartitionId()))
.ifPresent(fetchLogReqForBucket::setRoutingBucketCount);
}
return fetchLogRequests;
}

Map<Integer, FetchLogRequest> fetchLogRequests = new HashMap<>();
fetchReqsByLeader.forEach(
(leader, tablesReqs) -> {
FetchLogRequest request =
new FetchLogRequest()
.setFollowerServerId(-1)
.setMaxBytes(maxFetchBytes)
.setMinBytes(minFetchBytes)
.setMaxWaitMs(maxFetchWaitMs)
.setReadPreference(readPreference.value());
request.addAllTablesReqs(tablesReqs);
fetchLogRequests.put(leader, request);
});
return fetchLogRequests;
}

private List<TableBucket> fetchableBuckets() {
Expand Down
112 changes: 112 additions & 0 deletions ...s-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,12 +53,18 @@
import org.apache.fluss.rpc.gateway.CoordinatorGateway;
import org.apache.fluss.rpc.gateway.TabletServerGateway;
import org.apache.fluss.rpc.messages.ControlledShutdownRequest;
import org.apache.fluss.rpc.messages.FetchLogRequest;
import org.apache.fluss.rpc.messages.FetchLogResponse;
import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataRequest;
import org.apache.fluss.rpc.messages.InitWriterRequest;
import org.apache.fluss.rpc.messages.InitWriterResponse;
import org.apache.fluss.rpc.messages.MetadataRequest;
import org.apache.fluss.rpc.messages.PbFetchLogReqForBucket;
import org.apache.fluss.rpc.messages.PbFetchLogReqForTable;
import org.apache.fluss.rpc.messages.PbFetchLogRespForBucket;
import org.apache.fluss.rpc.messages.ReleaseKvSnapshotLeaseRequest;
import org.apache.fluss.rpc.metrics.TestingClientMetricGroup;
import org.apache.fluss.rpc.protocol.Errors;
import org.apache.fluss.security.acl.AccessControlEntry;
import org.apache.fluss.security.acl.AccessControlEntryFilter;
import org.apache.fluss.security.acl.AclBinding;
Expand All @@ -83,6 +89,7 @@
import org.junit.jupiter.params.provider.ValueSource;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
Expand All @@ -92,6 +99,7 @@
import java.util.concurrent.ExecutionException;

import static org.apache.fluss.config.ConfigOptions.DATALAKE_FORMAT;
import static org.apache.fluss.record.TestData.DATA1;
import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
import static org.apache.fluss.record.TestData.DATA1_SCHEMA_PK;
import static org.apache.fluss.record.TestData.DATA1_TABLE_DESCRIPTOR;
Expand All @@ -102,6 +110,8 @@
import static org.apache.fluss.security.acl.AccessControlEntry.WILD_CARD_HOST;
import static org.apache.fluss.security.acl.FlussPrincipal.WILD_CARD_PRINCIPAL;
import static org.apache.fluss.security.acl.OperationType.READ;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newProduceLogRequest;
import static org.apache.fluss.testutils.DataTestUtils.genMemoryLogRecordsByObject;
import static org.apache.fluss.testutils.DataTestUtils.row;
import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
import static org.assertj.core.api.Assertions.assertThat;
Expand Down Expand Up @@ -708,6 +718,108 @@ void testProduceWithNoWriteAuthorization() throws Exception {
}
}

@Test
void testFetchAuthorizationPreservesBucketOrder() throws Exception {
TablePath allowedPath = TablePath.of("test_db_1", "fetch_order_allowed");
rootAdmin
.createTable(
allowedPath,
TableDescriptor.builder().schema(DATA1_SCHEMA).distributedBy(9).build(),
false)
.get();
long allowedTableId = rootAdmin.getTableInfo(allowedPath).get().getTableId();
FLUSS_CLUSTER_EXTENSION.waitUntilTableReady(allowedTableId);

Map<Integer, List<TableBucket>> bucketsByLeader = new HashMap<>();
for (int bucketId = 0; bucketId < 9; bucketId++) {
TableBucket bucket = new TableBucket(allowedTableId, bucketId);
int leader = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(bucket);
bucketsByLeader.computeIfAbsent(leader, ignored -> new ArrayList<>()).add(bucket);
}
Map.Entry<Integer, List<TableBucket>> sameLeader =
bucketsByLeader.entrySet().stream()
.filter(entry -> entry.getValue().size() >= 2)
.findFirst()
.get();
TableBucket first = sameLeader.getValue().get(0);
TableBucket second = sameLeader.getValue().get(1);
FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(first);
FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(second);
TabletServerGateway rootGateway =
((FlussConnection) rootConn)
.getMetadataUpdater()
.newTabletServerClientForNode(sameLeader.getKey());
for (TableBucket bucket : Arrays.asList(first, second)) {
rootGateway
.produceLog(
newProduceLogRequest(
allowedTableId,
bucket.getBucket(),
-1,
genMemoryLogRecordsByObject(DATA1)))
.get();
}

List<AclBinding> readAcl =
Collections.singletonList(
new AclBinding(
Resource.table(allowedPath),
new AccessControlEntry(
guestPrincipal, "*", READ, PermissionType.ALLOW)));
rootAdmin.createAcls(readAcl).all().get();
FLUSS_CLUSTER_EXTENSION.waitUntilAuthenticationSync(readAcl, true);
TabletServerGateway guestGateway =
((FlussConnection) guestConn)
.getMetadataUpdater()
.newTabletServerClientForNode(sameLeader.getKey());
long deniedTableId = rootAdmin.getTableInfo(DATA1_TABLE_PATH_PK).get().getTableId();

for (TableBucket leading : Arrays.asList(first, second)) {
TableBucket trailing = leading.equals(first) ? second : first;
FetchLogRequest request = new FetchLogRequest().setFollowerServerId(-1).setMaxBytes(1);
request.addAllTablesReqs(
Arrays.asList(
fetchTableRequest(allowedTableId, leading.getBucket()),
fetchTableRequest(deniedTableId, 0),
fetchTableRequest(allowedTableId, trailing.getBucket())));
FetchLogResponse response = guestGateway.fetchLog(request).get();
try {
assertThat(fetchBucketResponse(response, leading).getRecordsSize()).isPositive();
assertThat(fetchBucketResponse(response, trailing).getRecordsSize()).isZero();
assertThat(
fetchBucketResponse(response, new TableBucket(deniedTableId, 0))
.getErrorCode())
.isEqualTo(Errors.AUTHORIZATION_EXCEPTION.code());
} finally {
if (response.getParsedByteBuf() != null) {
response.getParsedByteBuf().release();
}
}
}
}

private static PbFetchLogReqForTable fetchTableRequest(long tableId, int bucketId) {
PbFetchLogReqForTable request =
new PbFetchLogReqForTable().setTableId(tableId).setProjectionPushdownEnabled(false);
request.addAllBucketsReqs(
Collections.singletonList(
new PbFetchLogReqForBucket()
.setBucketId(bucketId)
.setFetchOffset(0L)
.setMaxFetchBytes(Integer.MAX_VALUE)));
return request;
}

private static PbFetchLogRespForBucket fetchBucketResponse(
FetchLogResponse response, TableBucket bucket) {
return response.getTablesRespsList().stream()
.filter(table -> table.getTableId() == bucket.getTableId())
.flatMap(table -> table.getBucketsRespsList().stream())
.filter(result -> result.getBucketId() == bucket.getBucket())
.findFirst()
.get();
}

@Test
void testProduceAndConsumer() throws Exception {
TableDescriptor descriptor =
Expand Down
Loading
Loading