Skip to content

Commit 33969e9

Browse files
luoyuxialuoyuxia
committed
resolve conflict
1 parent ccec817 commit 33969e9

File tree

10 files changed

+83
-76
lines changed

10 files changed

+83
-76
lines changed

‎fluss-common/src/main/java/org/apache/fluss/lake/committer/CommittedLakeSnapshot.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717

1818
package org.apache.fluss.lake.committer;
1919

20-
import java.util.HashMap;
2120
import java.util.Map;
2221
import java.util.Objects;
2322

@@ -30,10 +29,11 @@ public class CommittedLakeSnapshot {
3029
private final long lakeSnapshotId;
3130
// <partition_id, bucket> -> log offset, partition_id will be null if it's not a
3231
// partition bucket
33-
private final Map<String, String> lakeSnapshotProperties = new HashMap<>();
32+
private final Map<String, String> lakeSnapshotProperties;
3433

3534
public CommittedLakeSnapshot(long lakeSnapshotId, Map<String, String> lakeSnapshotProperties) {
3635
this.lakeSnapshotId = lakeSnapshotId;
36+
this.lakeSnapshotProperties = lakeSnapshotProperties;
3737
}
3838

3939
public long getLakeSnapshotId() {

‎fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@
3333
import org.apache.fluss.lake.writer.LakeTieringFactory;
3434
import org.apache.fluss.lake.writer.LakeWriter;
3535
import org.apache.fluss.metadata.TableBucket;
36-
import org.apache.fluss.metadata.TableInfo;
3736
import org.apache.fluss.metadata.TablePath;
3837
import org.apache.fluss.utils.ExceptionUtils;
3938

@@ -51,7 +50,6 @@
5150

5251
import java.util.ArrayList;
5352
import java.util.Collections;
54-
import java.util.Collections;
5553
import java.util.HashMap;
5654
import java.util.HashSet;
5755
import java.util.List;
@@ -202,7 +200,9 @@ private Committable commitWriteResults(
202200
}
203201
try (LakeCommitter<WriteResult, Committable> lakeCommitter =
204202
lakeTieringFactory.createLakeCommitter(
205-
new TieringCommitterInitContext(tablePath, admin.getTableInfo(tablePath).get(),
203+
new TieringCommitterInitContext(
204+
tablePath,
205+
admin.getTableInfo(tablePath).get(),
206206
lakeTieringConfig))) {
207207
List<WriteResult> writeResults =
208208
committableWriteResults.stream()

‎fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TestingLakeTieringFactory.java‎

Lines changed: 13 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.apache.fluss.lake.writer.LakeTieringFactory;
2727
import org.apache.fluss.lake.writer.LakeWriter;
2828
import org.apache.fluss.lake.writer.WriterInitContext;
29+
import org.apache.fluss.metadata.TablePath;
2930
import org.apache.fluss.record.LogRecord;
3031

3132
import javax.annotation.Nullable;
@@ -52,7 +53,7 @@ public TestingLakeTieringFactory() {
5253
@Override
5354
public LakeWriter<TestingWriteResult> createLakeWriter(WriterInitContext writerInitContext)
5455
throws IOException {
55-
return new TestingLakeWriter();
56+
return new TestingLakeWriter(writerInitContext.tablePath());
5657
}
5758

5859
@Override
@@ -77,8 +78,13 @@ public SimpleVersionedSerializer<TestingCommittable> getCommittableSerializer()
7778

7879
private static final class TestingLakeWriter implements LakeWriter<TestingWriteResult> {
7980

81+
private final TablePath tablePath;
8082
private int writtenRecords;
8183

84+
private TestingLakeWriter(TablePath tablePath) {
85+
this.tablePath = tablePath;
86+
}
87+
8288
@Override
8389
public void write(LogRecord record) throws IOException {
8490
writtenRecords += 1;
@@ -99,16 +105,14 @@ public static final class TestingLakeCommitter
99105

100106
private long currentSnapshot;
101107

102-
@Nullable private final CommittedLakeSnapshot mockCommittedSnapshot;
108+
private @Nullable CommittedLakeSnapshot missingCommittedLakeSnapshot;
103109

104110
public TestingLakeCommitter() {
105111
this(null);
106112
}
107113

108-
public TestingLakeCommitter(@Nullable CommittedLakeSnapshot mockCommittedSnapshot) {
109-
this.mockCommittedSnapshot = mockCommittedSnapshot;
110-
this.currentSnapshot =
111-
mockCommittedSnapshot == null ? 0 : mockCommittedSnapshot.getLakeSnapshotId();
114+
public TestingLakeCommitter(CommittedLakeSnapshot missingCommittedLakeSnapshot) {
115+
this.missingCommittedLakeSnapshot = missingCommittedLakeSnapshot;
112116
}
113117

114118
@Override
@@ -135,11 +139,10 @@ public void abort(TestingCommittable committable) throws IOException {
135139
@Override
136140
public @Nullable CommittedLakeSnapshot getMissingLakeSnapshot(
137141
@Nullable Long knownSnapshotId) throws IOException {
138-
if (knownSnapshotId == null) {
139-
return mockCommittedSnapshot;
140-
} else {
141-
return null;
142+
if (missingCommittedLakeSnapshot != null && knownSnapshotId == null) {
143+
return missingCommittedLakeSnapshot;
142144
}
145+
return null;
143146
}
144147

145148
@Override

‎fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java‎

Lines changed: 52 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -49,15 +49,12 @@
4949

5050
import javax.annotation.Nullable;
5151

52-
import java.io.IOException;
5352
import java.util.ArrayList;
5453
import java.util.Collections;
5554
import java.util.HashMap;
5655
import java.util.List;
5756
import java.util.Map;
5857

59-
import static org.apache.fluss.flink.tiering.committer.TieringCommitOperator.fromLogOffsetProperty;
60-
import static org.apache.fluss.flink.tiering.committer.TieringCommitOperator.toBucketOffsetsProperty;
6158
import static org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
6259
import static org.apache.fluss.record.TestData.DATA1_PARTITIONED_TABLE_DESCRIPTOR;
6360
import static org.assertj.core.api.Assertions.assertThat;
@@ -181,7 +178,6 @@ void testCommitPartitionedTable() throws Exception {
181178
Map<String, Long> partitionIdByNames =
182179
FLUSS_CLUSTER_EXTENSION.waitUntilPartitionAllReady(tablePath);
183180
Map<TableBucket, Long> expectedLogEndOffsets = new HashMap<>();
184-
Map<TableBucket, Long> expectedMaxTimestamps = new HashMap<>();
185181
int numberOfWriteResults = 3 * partitionIdByNames.size();
186182
long offset = 0;
187183
long timestamp = System.currentTimeMillis();
@@ -201,7 +197,6 @@ void testCommitPartitionedTable() throws Exception {
201197
currentTimestamp,
202198
numberOfWriteResults));
203199
expectedLogEndOffsets.put(tableBucket, currentOffset);
204-
expectedMaxTimestamps.put(tableBucket, currentTimestamp);
205200
}
206201
if (bucket == 2) {
207202
verifyLakeSnapshot(tablePath, tableId, 1, expectedLogEndOffsets);
@@ -261,10 +256,17 @@ void testTableCommitWhenFlussMissingLakeSnapshot() throws Exception {
261256
long tableId = createTable(tablePath, DEFAULT_PK_TABLE_DESCRIPTOR);
262257
int numberOfWriteResults = 3;
263258

264-
CommittedLakeSnapshot mockCommittedSnapshot =
265-
mockCommittedLakeSnapshot(Collections.singletonList(null), tableId, 2);
259+
Map<TableBucket, Long> expectedLogEndOffsets = new HashMap<>();
260+
for (int bucket = 0; bucket < 3; bucket++) {
261+
TableBucket tableBucket = new TableBucket(tableId, bucket);
262+
expectedLogEndOffsets.put(tableBucket, 3L);
263+
}
264+
265+
CommittedLakeSnapshot mockMissingCommittedLakeSnapshot =
266+
mockCommittedLakeSnapshot(tableId, tablePath, 0, expectedLogEndOffsets);
266267
TestingLakeTieringFactory.TestingLakeCommitter testingLakeCommitter =
267-
new TestingLakeTieringFactory.TestingLakeCommitter(mockCommittedSnapshot);
268+
new TestingLakeTieringFactory.TestingLakeCommitter(
269+
mockMissingCommittedLakeSnapshot);
268270
committerOperator =
269271
new TieringCommitOperator<>(
270272
parameters,
@@ -283,18 +285,18 @@ void testTableCommitWhenFlussMissingLakeSnapshot() throws Exception {
283285
verifyLakeSnapshot(
284286
tablePath,
285287
tableId,
286-
2,
287-
getExpectedLogEndOffsets(tableId, mockCommittedSnapshot),
288+
0,
289+
expectedLogEndOffsets,
288290
String.format(
289291
"The current Fluss's lake snapshot %s is less than lake actual snapshot %d committed by Fluss for table: {tablePath=%s, tableId=%d},"
290292
+ " missing snapshot: %s.",
291293
null,
292-
mockCommittedSnapshot.getLakeSnapshotId(),
294+
mockMissingCommittedLakeSnapshot.getLakeSnapshotId(),
293295
tablePath,
294296
tableId,
295-
mockCommittedSnapshot));
297+
mockMissingCommittedLakeSnapshot));
296298

297-
Map<TableBucket, Long> expectedLogEndOffsets = new HashMap<>();
299+
expectedLogEndOffsets = new HashMap<>();
298300
for (int bucket = 0; bucket < 3; bucket++) {
299301
TableBucket tableBucket = new TableBucket(tableId, bucket);
300302
long offset = bucket * bucket;
@@ -305,7 +307,7 @@ void testTableCommitWhenFlussMissingLakeSnapshot() throws Exception {
305307
expectedLogEndOffsets.put(tableBucket, offset);
306308
}
307309

308-
verifyLakeSnapshot(tablePath, tableId, 3, expectedLogEndOffsets);
310+
verifyLakeSnapshot(tablePath, tableId, 1, expectedLogEndOffsets);
309311
}
310312

311313
@Test
@@ -318,10 +320,21 @@ void testPartitionedTableCommitWhenFlussMissingLakeSnapshot() throws Exception {
318320
Map<String, Long> partitionIdByNames =
319321
FLUSS_CLUSTER_EXTENSION.waitUntilPartitionAllReady(tablePath);
320322

321-
CommittedLakeSnapshot mockCommittedSnapshot =
322-
mockCommittedLakeSnapshot(Collections.singletonList(null), tableId, 3);
323+
Map<TableBucket, Long> expectedLogEndOffsets = new HashMap<>();
324+
for (int bucket = 0; bucket < 3; bucket++) {
325+
for (String partitionName : partitionIdByNames.keySet()) {
326+
long partitionId = partitionIdByNames.get(partitionName);
327+
TableBucket tableBucket = new TableBucket(tableId, partitionId, bucket);
328+
expectedLogEndOffsets.put(tableBucket, 3L);
329+
}
330+
}
331+
332+
CommittedLakeSnapshot mockMissingCommittedLakeSnapshot =
333+
mockCommittedLakeSnapshot(tableId, tablePath, 0, expectedLogEndOffsets);
334+
323335
TestingLakeTieringFactory.TestingLakeCommitter testingLakeCommitter =
324-
new TestingLakeTieringFactory.TestingLakeCommitter(mockCommittedSnapshot);
336+
new TestingLakeTieringFactory.TestingLakeCommitter(
337+
mockMissingCommittedLakeSnapshot);
325338
committerOperator =
326339
new TieringCommitOperator<>(
327340
parameters,
@@ -351,35 +364,15 @@ void testPartitionedTableCommitWhenFlussMissingLakeSnapshot() throws Exception {
351364
verifyLakeSnapshot(
352365
tablePath,
353366
tableId,
354-
3,
355-
getExpectedLogEndOffsets(tableId, mockCommittedSnapshot),
367+
0,
368+
expectedLogEndOffsets,
356369
String.format(
357370
"The current Fluss's lake snapshot %s is less than lake actual snapshot %d committed by Fluss for table: {tablePath=%s, tableId=%d}, missing snapshot: %s.",
358371
null,
359-
mockCommittedSnapshot.getLakeSnapshotId(),
372+
mockMissingCommittedLakeSnapshot.getLakeSnapshotId(),
360373
tablePath,
361374
tableId,
362-
mockCommittedSnapshot));
363-
}
364-
365-
private CommittedLakeSnapshot mockCommittedLakeSnapshot(
366-
List<Long> partitions, long tableId, int snapshotId) throws IOException {
367-
Map<TableBucket, Long> logEndOffsets = new HashMap<>();
368-
for (Long partition : partitions) {
369-
for (int bucket = 0; bucket < DEFAULT_BUCKET_NUM; bucket++) {
370-
logEndOffsets.put(new TableBucket(tableId, partition, bucket), bucket + 1L);
371-
}
372-
}
373-
return new CommittedLakeSnapshot(snapshotId, toBucketOffsetsProperty(logEndOffsets));
374-
}
375-
376-
private Map<TableBucket, Long> getExpectedLogEndOffsets(
377-
long tableId, CommittedLakeSnapshot committedLakeSnapshot) throws IOException {
378-
return fromLogOffsetProperty(
379-
tableId,
380-
committedLakeSnapshot
381-
.getSnapshotProperties()
382-
.get(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY));
375+
mockMissingCommittedLakeSnapshot));
383376
}
384377

385378
private StreamRecord<TableBucketWriteResult<TestingWriteResult>>
@@ -461,10 +454,25 @@ private void verifyLakeSnapshot(
461454
List<OperatorEvent> operatorEvents = mockOperatorEventGateway.getEventsSent();
462455
SourceEventWrapper sourceEventWrapper =
463456
(SourceEventWrapper) operatorEvents.get(operatorEvents.size() - 1);
464-
FailedTieringEvent finishTieringEvent =
457+
FailedTieringEvent failedTieringEvent =
465458
(FailedTieringEvent) sourceEventWrapper.getSourceEvent();
466-
assertThat(finishTieringEvent.getTableId()).isEqualTo(tableId);
467-
assertThat(finishTieringEvent.failReason()).contains(failedReason);
459+
assertThat(failedTieringEvent.getTableId()).isEqualTo(tableId);
460+
assertThat(failedTieringEvent.failReason()).contains(failedReason);
461+
}
462+
463+
private CommittedLakeSnapshot mockCommittedLakeSnapshot(
464+
long tableId, TablePath tablePath, int snapshotId, Map<TableBucket, Long> logEndOffsets)
465+
throws Exception {
466+
try (FlussTableLakeSnapshotCommitter lakeSnapshotCommitter =
467+
new FlussTableLakeSnapshotCommitter(FLUSS_CLUSTER_EXTENSION.getClientConfig())) {
468+
lakeSnapshotCommitter.open();
469+
String lakeSnapshotFile =
470+
lakeSnapshotCommitter.prepareCommit(tableId, tablePath, logEndOffsets);
471+
return new CommittedLakeSnapshot(
472+
snapshotId,
473+
Collections.singletonMap(
474+
FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, lakeSnapshotFile));
475+
}
468476
}
469477

470478
private static class MockOperatorEventDispatcher implements OperatorEventDispatcher {

‎fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringTest.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -64,8 +64,6 @@
6464
import java.util.Map;
6565
import java.util.stream.Stream;
6666

67-
import static org.apache.fluss.flink.tiering.committer.TieringCommitOperator.fromLogOffsetProperty;
68-
import static org.apache.fluss.flink.tiering.committer.TieringCommitOperator.toBucketOffsetsProperty;
6967
import static org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
7068
import static org.assertj.core.api.Assertions.assertThat;
7169

‎fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java‎

Lines changed: 7 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
import org.apache.fluss.row.BinaryString;
3636
import org.apache.fluss.row.GenericRow;
3737
import org.apache.fluss.utils.types.Tuple2;
38+
3839
import org.apache.paimon.CoreOptions;
3940
import org.apache.paimon.catalog.Catalog;
4041
import org.apache.paimon.catalog.CatalogContext;
@@ -58,6 +59,7 @@
5859
import org.junit.jupiter.params.provider.MethodSource;
5960

6061
import javax.annotation.Nullable;
62+
6163
import java.io.File;
6264
import java.io.IOException;
6365
import java.util.ArrayList;
@@ -69,8 +71,6 @@
6971
import java.util.Map;
7072
import java.util.stream.Stream;
7173

72-
import static org.apache.fluss.flink.tiering.committer.TieringCommitOperator.fromLogOffsetProperty;
73-
import static org.apache.fluss.flink.tiering.committer.TieringCommitOperator.toBucketOffsetsProperty;
7474
import static org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
7575
import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon;
7676
import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
@@ -188,16 +188,7 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitioned) thr
188188
CommittedLakeSnapshot committedLakeSnapshot = lakeCommitter.getMissingLakeSnapshot(0L);
189189
assertThat(committedLakeSnapshot).isNotNull();
190190
Map<String, String> snapshotProperty = new HashMap<>();
191-
// todo:
192-
// Map<Tuple2<Long, Integer>, Long> offsets =
193-
// committedLakeSnapshot.getLogEndOffsets();
194-
Map<Tuple2<Long, Integer>, Long> offsets = new HashMap<>();
195-
for (int bucket = 0; bucket < 3; bucket++) {
196-
for (Long partitionId : partitionIdAndName.keySet()) {
197-
// we only write 10 records, so expected log offset should be 10
198-
assertThat(offsets.get(Tuple2.of(partitionId, bucket))).isEqualTo(10);
199-
}
200-
}
191+
201192
assertThat(committedLakeSnapshot.getLakeSnapshotId()).isOne();
202193

203194
// use null as the known snapshot id
@@ -875,7 +866,6 @@ private void writeData(
875866
SimpleVersionedSerializer<PaimonCommittable> committableSerializer =
876867
paimonLakeTieringFactory.getCommittableSerializer();
877868

878-
Map<TableBucket, Long> tableBucketOffsets = new HashMap<>();
879869
// first, write data
880870
for (int bucket = 0; bucket < bucketNum; bucket++) {
881871
for (Map.Entry<Long, String> entry : partitionIdAndName.entrySet()) {
@@ -890,7 +880,6 @@ private void writeData(
890880
List<LogRecord> writtenRecords = writeAndExpectRecords.f0;
891881
List<LogRecord> expectRecords = writeAndExpectRecords.f1;
892882
recordsByBucket.put(partitionBucket, expectRecords);
893-
tableBucketOffsets.put(new TableBucket(0, entry.getKey(), bucket), 10L);
894883
for (LogRecord logRecord : writtenRecords) {
895884
lakeWriter.write(logRecord);
896885
}
@@ -913,7 +902,10 @@ private void writeData(
913902
paimonCommittable =
914903
committableSerializer.deserialize(
915904
committableSerializer.getVersion(), serialized);
916-
lakeCommitter.commit(paimonCommittable, toBucketOffsetsProperty(tableBucketOffsets));
905+
906+
Map<String, String> snapshotProperties =
907+
Collections.singletonMap(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, "offsets");
908+
lakeCommitter.commit(paimonCommittable, snapshotProperties);
917909
}
918910
}
919911
}

‎fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,10 +103,12 @@
103103
import org.apache.fluss.server.zk.data.BucketSnapshot;
104104
import org.apache.fluss.server.zk.data.lake.LakeTableHelper;
105105
import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot;
106+
106107
import org.slf4j.Logger;
107108
import org.slf4j.LoggerFactory;
108109

109110
import javax.annotation.Nullable;
111+
110112
import java.util.ArrayList;
111113
import java.util.Collection;
112114
import java.util.HashSet;

0 commit comments

Comments
 (0)