Skip to content

Commit 52480eb

Browse files
gyang94gyang94
authored
[server] Complete delayed fetches after log appends (#3412)
1 parent c07dcc6 commit 52480eb

2 files changed

Lines changed: 279 additions & 6 deletions

File tree

‎fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -721,9 +721,11 @@ public void appendRecordsToLog(
721721
appendToLocalLog(entriesPerBucket, requiredAcks, userContext);
722722
LOG.debug("Append records to local log in {} ms", System.currentTimeMillis() - startTime);
723723

724-
// maybe do delay write operation.
724+
// Register any delayed write before waking followers. The append has already released
725+
// the replica locks, so completion checks can run directly here.
725726
maybeAddDelayedWrite(
726727
timeoutMs, requiredAcks, entriesPerBucket.size(), appendResult, responseCallback);
728+
completeDelayedFetches(appendResult);
727729
}
728730

729731
/** Appends historical log batches while preserving each original partition in the response. */
@@ -2236,6 +2238,23 @@ private boolean isNonCriticalFetchError(Errors error) {
22362238
|| error == Errors.UNKNOWN_TABLE_OR_BUCKET_EXCEPTION;
22372239
}
22382240

2241+
private void completeDelayedFetches(Map<TableBucket, ProduceLogResultForBucket> appendResults) {
2242+
appendResults.forEach(
2243+
(tableBucket, appendResult) -> {
2244+
if (appendResult.succeeded()) {
2245+
try {
2246+
delayedFetchLogManager.checkAndComplete(
2247+
new DelayedTableBucketKey(tableBucket));
2248+
} catch (Exception e) {
2249+
LOG.error(
2250+
"Failed to complete delayed fetches for bucket {}.",
2251+
tableBucket,
2252+
e);
2253+
}
2254+
}
2255+
});
2256+
}
2257+
22392258
private void completeDelayedOperations(TableBucket tableBucket) {
22402259
DelayedTableBucketKey delayedTableBucketKey = new DelayedTableBucketKey(tableBucket);
22412260
delayedWriteManager.checkAndComplete(delayedTableBucketKey);

‎fluss-server/src/test/java/org/apache/fluss/server/replica/delay/DelayedFetchLogTest.java‎

Lines changed: 259 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -25,16 +25,20 @@
2525
import org.apache.fluss.server.log.FetchParams;
2626
import org.apache.fluss.server.log.LogOffsetMetadata;
2727
import org.apache.fluss.server.metrics.group.TestingMetricGroups;
28+
import org.apache.fluss.server.replica.Replica;
2829
import org.apache.fluss.server.replica.ReplicaTestBase;
2930
import org.apache.fluss.server.replica.delay.DelayedFetchLog.FetchBucketStatus;
3031

3132
import org.junit.jupiter.api.Test;
3233

3334
import java.time.Duration;
35+
import java.util.Arrays;
3436
import java.util.Collections;
37+
import java.util.HashMap;
3538
import java.util.List;
3639
import java.util.Map;
3740
import java.util.concurrent.CompletableFuture;
41+
import java.util.concurrent.atomic.AtomicInteger;
3842
import java.util.function.Consumer;
3943

4044
import static org.apache.fluss.record.TestData.DATA1;
@@ -83,7 +87,7 @@ void testCompleteDelayedFetchLog() throws Exception {
8387
assertThat(delayedFetchLogManager.numDelayed()).isEqualTo(1);
8488
assertThat(delayedFetchLogManager.watched()).isEqualTo(1);
8589

86-
// write data.
90+
// Appending data completes the fetch without an additional completion trigger.
8791
assertThat(delayedResponse.isDone()).isFalse();
8892
CompletableFuture<List<ProduceLogResultForBucket>> future = new CompletableFuture<>();
8993
replicaManager.appendRecordsToLog(
@@ -93,10 +97,7 @@ void testCompleteDelayedFetchLog() throws Exception {
9397
null,
9498
future::complete);
9599
assertThat(future.get()).containsOnly(new ProduceLogResultForBucket(tb, 0, 10L));
96-
97-
// check and complete manually
98-
numComplete = delayedFetchLogManager.checkAndComplete(delayedTableBucketKey);
99-
assertThat(numComplete).isEqualTo(1);
100+
assertThat(delayedResponse.isDone()).isTrue();
100101
assertThat(delayedFetchLogManager.numDelayed()).isEqualTo(0);
101102
assertThat(delayedFetchLogManager.watched()).isEqualTo(0);
102103

@@ -107,6 +108,186 @@ void testCompleteDelayedFetchLog() throws Exception {
107108
assertLogRecordsEquals(DATA1_ROW_TYPE, resultForBucket.records(), DATA1);
108109
}
109110

111+
@Test
112+
void testSuccessfulBucketCompletesWhenAnotherBucketAppendFails() throws Exception {
113+
TableBucket successfulBucket = new TableBucket(DATA1_TABLE_ID, 1);
114+
TableBucket failedBucket = new TableBucket(DATA1_TABLE_ID, 2);
115+
makeLogTableAsLeader(successfulBucket.getBucket());
116+
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> delayedResponse =
117+
watchDelayedFetch(successfulBucket);
118+
119+
Map<TableBucket, MemoryLogRecords> entries = new HashMap<>();
120+
entries.put(successfulBucket, genMemoryLogRecordsByObject(DATA1));
121+
entries.put(failedBucket, genMemoryLogRecordsByObject(DATA1));
122+
CompletableFuture<List<ProduceLogResultForBucket>> produceResponse =
123+
new CompletableFuture<>();
124+
125+
replicaManager.appendRecordsToLog(20000, 1, entries, null, produceResponse::complete);
126+
127+
List<ProduceLogResultForBucket> produceResults = produceResponse.get();
128+
assertThat(produceResults).hasSize(2);
129+
assertThat(produceResults)
130+
.filteredOn(result -> result.getTableBucket().equals(successfulBucket))
131+
.hasSize(1)
132+
.allSatisfy(result -> assertThat(result.succeeded()).isTrue());
133+
assertThat(produceResults)
134+
.filteredOn(result -> result.getTableBucket().equals(failedBucket))
135+
.hasSize(1)
136+
.allSatisfy(result -> assertThat(result.failed()).isTrue());
137+
assertThat(delayedResponse).isDone();
138+
assertThat(replicaManager.getDelayedFetchLogManager().numDelayed()).isZero();
139+
}
140+
141+
@Test
142+
void testAppendCompletesDelayedFetchesForMultipleBuckets() throws Exception {
143+
TableBucket firstBucket = new TableBucket(DATA1_TABLE_ID, 1);
144+
TableBucket secondBucket = new TableBucket(DATA1_TABLE_ID, 2);
145+
makeLogTableAsLeader(firstBucket.getBucket());
146+
makeLogTableAsLeader(secondBucket.getBucket());
147+
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> firstResponse =
148+
watchDelayedFetch(firstBucket);
149+
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> secondResponse =
150+
watchDelayedFetch(secondBucket);
151+
152+
Map<TableBucket, MemoryLogRecords> entries = new HashMap<>();
153+
entries.put(firstBucket, genMemoryLogRecordsByObject(DATA1));
154+
entries.put(secondBucket, genMemoryLogRecordsByObject(DATA1));
155+
CompletableFuture<List<ProduceLogResultForBucket>> produceResponse =
156+
new CompletableFuture<>();
157+
158+
replicaManager.appendRecordsToLog(20000, 1, entries, null, produceResponse::complete);
159+
160+
assertThat(produceResponse.get())
161+
.hasSize(2)
162+
.allSatisfy(result -> assertThat(result.succeeded()).isTrue());
163+
assertThat(firstResponse).isDone();
164+
assertThat(secondResponse).isDone();
165+
assertThat(replicaManager.getDelayedFetchLogManager().numDelayed()).isZero();
166+
}
167+
168+
@Test
169+
void testDelayedFollowerFetchesWithAllAcks() throws Exception {
170+
TableBucket tableBucket = new TableBucket(DATA1_TABLE_ID, 1);
171+
int firstFollowerId = 2;
172+
int secondFollowerId = 3;
173+
List<Integer> replicas = Arrays.asList(TABLET_SERVER_ID, firstFollowerId, secondFollowerId);
174+
makeLogTableAsLeader(tableBucket, replicas, replicas, false);
175+
Replica leader = replicaManager.getReplicaOrException(tableBucket);
176+
long logEndOffset = DATA1.size();
177+
178+
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> firstFollowerFetch =
179+
fetchLog(tableBucket, firstFollowerId, 0L, 1);
180+
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> secondFollowerFetch =
181+
fetchLog(tableBucket, secondFollowerId, 0L, 1);
182+
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> consumerFetch =
183+
fetchLog(tableBucket, -1, 0L, 1);
184+
assertThat(firstFollowerFetch).isNotDone();
185+
assertThat(secondFollowerFetch).isNotDone();
186+
assertThat(consumerFetch).isNotDone();
187+
assertThat(replicaManager.getDelayedFetchLogManager().numDelayed()).isEqualTo(3);
188+
189+
CompletableFuture<List<ProduceLogResultForBucket>> produceResponse =
190+
new CompletableFuture<>();
191+
replicaManager.appendRecordsToLog(
192+
(int) Duration.ofMinutes(3).toMillis(),
193+
-1,
194+
Collections.singletonMap(tableBucket, genMemoryLogRecordsByObject(DATA1)),
195+
null,
196+
produceResponse::complete);
197+
198+
assertThat(leader.getLogTablet().localLogEndOffset()).isEqualTo(logEndOffset);
199+
assertThat(leader.getLogHighWatermark()).isZero();
200+
assertThat(produceResponse).isNotDone();
201+
assertThat(consumerFetch).isNotDone();
202+
assertThat(replicaManager.getDelayedWriteManager().numDelayed()).isOne();
203+
204+
// Followers read up to LEO, while the consumer and acks=all produce must wait for HW.
205+
assertFetchedRecords(firstFollowerFetch, tableBucket, 0L);
206+
assertFetchedRecords(secondFollowerFetch, tableBucket, 0L);
207+
assertThat(leader.getLogHighWatermark()).isZero();
208+
assertThat(produceResponse).isNotDone();
209+
assertThat(consumerFetch).isNotDone();
210+
assertThat(replicaManager.getDelayedFetchLogManager().numDelayed()).isOne();
211+
212+
// Fetching from the new LEO acknowledges the records returned by the previous fetch.
213+
acknowledgeFetchedRecords(tableBucket, firstFollowerId, logEndOffset);
214+
assertThat(leader.getLogHighWatermark()).isZero();
215+
assertThat(produceResponse).isNotDone();
216+
assertThat(consumerFetch).isNotDone();
217+
218+
acknowledgeFetchedRecords(tableBucket, secondFollowerId, logEndOffset);
219+
assertThat(leader.getLogHighWatermark()).isEqualTo(logEndOffset);
220+
assertThat(produceResponse).isDone();
221+
assertThat(produceResponse.join())
222+
.containsExactly(new ProduceLogResultForBucket(tableBucket, 0L, logEndOffset));
223+
assertFetchedRecords(consumerFetch, tableBucket, logEndOffset);
224+
assertThat(replicaManager.getDelayedWriteManager().numDelayed()).isZero();
225+
assertThat(replicaManager.getDelayedFetchLogManager().numDelayed()).isZero();
226+
}
227+
228+
@Test
229+
void testFetchCompletionFailureDoesNotFailProduceOrOtherBuckets() throws Exception {
230+
TableBucket firstBucket = new TableBucket(DATA1_TABLE_ID, 1);
231+
TableBucket secondBucket = new TableBucket(DATA1_TABLE_ID, 2);
232+
int followerId = 2;
233+
List<Integer> replicas = Arrays.asList(TABLET_SERVER_ID, followerId);
234+
Map<TableBucket, MemoryLogRecords> entries = new HashMap<>();
235+
Map<TableBucket, FetchLogResultForBucket> fetchResults = new HashMap<>();
236+
AtomicInteger completedFetches = new AtomicInteger();
237+
AtomicInteger pendingWritesAtFirstFetch = new AtomicInteger(-1);
238+
for (TableBucket tableBucket : Arrays.asList(firstBucket, secondBucket)) {
239+
makeLogTableAsLeader(tableBucket, replicas, replicas, false);
240+
entries.put(tableBucket, genMemoryLogRecordsByObject(DATA1));
241+
fetchLog(
242+
tableBucket,
243+
followerId,
244+
0L,
245+
1,
246+
results -> {
247+
fetchResults.putAll(results);
248+
// Fail the first callback regardless of bucket iteration order.
249+
if (completedFetches.getAndIncrement() == 0) {
250+
pendingWritesAtFirstFetch.set(
251+
replicaManager.getDelayedWriteManager().numDelayed());
252+
throw new IllegalStateException("Expected fetch callback failure");
253+
}
254+
});
255+
}
256+
assertThat(completedFetches).hasValue(0);
257+
assertThat(replicaManager.getDelayedFetchLogManager().numDelayed()).isEqualTo(2);
258+
CompletableFuture<List<ProduceLogResultForBucket>> produceResponse =
259+
new CompletableFuture<>();
260+
261+
replicaManager.appendRecordsToLog(
262+
(int) Duration.ofMinutes(3).toMillis(),
263+
-1,
264+
entries,
265+
null,
266+
produceResponse::complete);
267+
268+
assertThat(completedFetches).hasValue(2);
269+
assertThat(pendingWritesAtFirstFetch).hasValue(1);
270+
assertThat(fetchResults).containsOnlyKeys(firstBucket, secondBucket);
271+
for (FetchLogResultForBucket result : fetchResults.values()) {
272+
assertThat(result.succeeded()).isTrue();
273+
assertThat(result.getHighWatermark()).isZero();
274+
assertLogRecordsEquals(DATA1_ROW_TYPE, result.records(), DATA1);
275+
}
276+
assertThat(produceResponse).isNotDone();
277+
assertThat(replicaManager.getDelayedWriteManager().numDelayed()).isOne();
278+
assertThat(replicaManager.getDelayedFetchLogManager().numDelayed()).isZero();
279+
280+
acknowledgeFetchedRecords(firstBucket, followerId, DATA1.size());
281+
assertThat(produceResponse).isNotDone();
282+
acknowledgeFetchedRecords(secondBucket, followerId, DATA1.size());
283+
assertThat(produceResponse).isDone();
284+
assertThat(produceResponse.join())
285+
.containsExactlyInAnyOrder(
286+
new ProduceLogResultForBucket(firstBucket, 0L, DATA1.size()),
287+
new ProduceLogResultForBucket(secondBucket, 0L, DATA1.size()));
288+
assertThat(replicaManager.getDelayedWriteManager().numDelayed()).isZero();
289+
}
290+
110291
@Test
111292
void testDelayFetchLogTimeout() {
112293
TableBucket tb = new TableBucket(DATA1_TABLE_ID, 1);
@@ -150,6 +331,55 @@ void testDelayFetchLogTimeout() {
150331
});
151332
}
152333

334+
private CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> fetchLog(
335+
TableBucket tableBucket, int replicaId, long fetchOffset, int minFetchBytes) {
336+
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> response =
337+
new CompletableFuture<>();
338+
fetchLog(tableBucket, replicaId, fetchOffset, minFetchBytes, response::complete);
339+
return response;
340+
}
341+
342+
private void fetchLog(
343+
TableBucket tableBucket,
344+
int replicaId,
345+
long fetchOffset,
346+
int minFetchBytes,
347+
Consumer<Map<TableBucket, FetchLogResultForBucket>> responseCallback) {
348+
replicaManager.fetchLogRecords(
349+
new FetchParams(
350+
replicaId,
351+
Integer.MAX_VALUE,
352+
minFetchBytes,
353+
Duration.ofMinutes(3).toMillis()),
354+
Collections.singletonMap(
355+
tableBucket,
356+
new FetchReqInfo(tableBucket.getTableId(), fetchOffset, Integer.MAX_VALUE)),
357+
null,
358+
responseCallback);
359+
}
360+
361+
private void acknowledgeFetchedRecords(TableBucket tableBucket, int followerId, long offset) {
362+
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> response =
363+
fetchLog(tableBucket, followerId, offset, 0);
364+
assertThat(response).isDone();
365+
FetchLogResultForBucket result = response.join().get(tableBucket);
366+
assertThat(result.succeeded()).isTrue();
367+
assertThat(result.recordsOrEmpty().sizeInBytes()).isZero();
368+
}
369+
370+
private void assertFetchedRecords(
371+
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> response,
372+
TableBucket tableBucket,
373+
long expectedHighWatermark)
374+
throws Exception {
375+
assertThat(response).isDone();
376+
assertThat(response.join()).containsOnlyKeys(tableBucket);
377+
FetchLogResultForBucket result = response.join().get(tableBucket);
378+
assertThat(result.succeeded()).isTrue();
379+
assertThat(result.getHighWatermark()).isEqualTo(expectedHighWatermark);
380+
assertLogRecordsEquals(DATA1_ROW_TYPE, result.records(), DATA1);
381+
}
382+
153383
private DelayedFetchLog createDelayedFetchLogRequest(
154384
TableBucket tb,
155385
int minFetchSize,
@@ -165,4 +395,28 @@ private DelayedFetchLog createDelayedFetchLogRequest(
165395
TestingMetricGroups.TABLET_SERVER_METRICS,
166396
null);
167397
}
398+
399+
private CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> watchDelayedFetch(
400+
TableBucket tableBucket) {
401+
FetchLogResultForBucket previousResult =
402+
FetchLogResultForBucket.records(tableBucket, MemoryLogRecords.EMPTY, 0L, -1L, -1L);
403+
CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> response =
404+
new CompletableFuture<>();
405+
DelayedFetchLog delayedFetchLog =
406+
createDelayedFetchLogRequest(
407+
tableBucket,
408+
1,
409+
Duration.ofMinutes(3).toMillis(),
410+
new FetchBucketStatus(
411+
new FetchReqInfo(150001L, 0L, Integer.MAX_VALUE),
412+
new LogOffsetMetadata(0L, 0L, 0),
413+
previousResult),
414+
response::complete);
415+
replicaManager
416+
.getDelayedFetchLogManager()
417+
.tryCompleteElseWatch(
418+
delayedFetchLog,
419+
Collections.singletonList(new DelayedTableBucketKey(tableBucket)));
420+
return response;
421+
}
168422
}

0 commit comments

Comments
 (0)