You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2025/12/01 12:32:44 UTC
(fluss) branch ci-add-column updated (8660164b0 -> 1e3ecf289)
This is an automated email from the ASF dual-hosted git repository.
jark pushed a change to branch ci-add-column
in repository https://gitbox.apache.org/repos/asf/fluss.git
from 8660164b0 KvTablet
new 57166d866 ServerSchemaCache TablePath -> tableId
new 639c49203 WIP
new 1e3ecf289 WIP
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../java/org/apache/fluss/record/BinaryValue.java | 9 +--
.../org/apache/fluss/row/encode/ValueDecoder.java | 12 ----
.../org/apache/fluss/row/encode/ValueEncoder.java | 24 -------
.../java/org/apache/fluss/row/PaddingRowTest.java | 78 ++++++++++++++++++++++
.../org/apache/fluss/server/RpcServiceBase.java | 2 +-
.../fluss/server/coordinator/MetadataManager.java | 4 +-
.../org/apache/fluss/server/log/FetchParams.java | 2 +-
.../java/org/apache/fluss/server/log/LocalLog.java | 2 -
.../server/log/remote/FsRemoteLogOutputStream.java | 2 +-
.../fluss/server/log/remote/LogSegmentFiles.java | 2 +-
.../fluss/server/log/remote/LogTieringTask.java | 4 +-
.../fluss/server/metadata/ServerSchemaCache.java | 74 ++++++++++----------
.../server/metadata/TabletServerMetadataCache.java | 13 ++--
.../fluss/server/replica/AdjustIsrManager.java | 2 +-
.../org/apache/fluss/server/replica/Replica.java | 5 +-
.../replica/fetcher/ReplicaFetcherManager.java | 2 +-
.../replica/fetcher/ReplicaFetcherThread.java | 2 +-
.../fluss/server/utils/RpcGatewayManager.java | 2 +-
.../fluss/server/utils/ServerRpcMessageUtils.java | 2 +-
.../org/apache/fluss/server/utils/timer/Timer.java | 2 +-
.../fluss/server/utils/timer/TimingWheel.java | 4 +-
.../apache/fluss/server/zk/ZooKeeperClient.java | 50 +++++++++++---
.../server/authorizer/DefaultAuthorizerTest.java | 4 +-
.../CompletedSnapshotStoreManagerTest.java | 8 +--
.../coordinator/CoordinatorChannelManagerTest.java | 6 +-
.../fluss/server/coordinator/TableManagerTest.java | 2 -
.../server/kv/prewrite/KvPreWriteBufferTest.java | 4 +-
.../kv/snapshot/CompletedSnapshotStoreTest.java | 2 +-
.../kv/snapshot/KvTabletSnapshotTargetTest.java | 2 +-
.../ZooKeeperCompletedSnapshotStoreTest.java | 4 +-
.../apache/fluss/server/log/LogManagerTest.java | 2 +-
.../org/apache/fluss/server/log/LogTabletTest.java | 4 +-
.../server/log/remote/RemoteLogTabletTest.java | 4 +-
.../server/metadata/MetadataUpdateITCase.java | 2 +-
.../server/metadata/ServerSchemaCacheTest.java | 46 ++++++-------
.../fluss/server/replica/AdjustIsrITCase.java | 2 +-
.../fluss/server/replica/AdjustIsrManagerTest.java | 2 +-
.../replica/HighWatermarkPersistenceTest.java | 2 +-
.../replica/RemoveOfflineReplicaFromIsrITCase.java | 2 +-
.../fluss/server/replica/ReplicaManagerTest.java | 16 +++--
.../apache/fluss/server/replica/ReplicaTest.java | 10 +--
.../fluss/server/replica/ReplicaTestBase.java | 4 +-
.../fluss/server/tablet/TabletServiceITCase.java | 4 +-
.../server/testutils/FlussClusterExtension.java | 16 ++---
fluss-test-coverage/pom.xml | 4 +-
45 files changed, 257 insertions(+), 194 deletions(-)
create mode 100644 fluss-common/src/test/java/org/apache/fluss/row/PaddingRowTest.java
(fluss) 01/03: ServerSchemaCache TablePath -> tableId
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch ci-add-column
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 57166d866e427c79a4e133095442fb261e6fbdd1
Author: Jark Wu <ja...@apache.org>
AuthorDate: Mon Dec 1 19:19:48 2025 +0800
ServerSchemaCache TablePath -> tableId
---
.../org/apache/fluss/server/log/FetchParams.java | 2 +-
.../java/org/apache/fluss/server/log/LocalLog.java | 2 -
.../server/log/remote/FsRemoteLogOutputStream.java | 2 +-
.../fluss/server/log/remote/LogSegmentFiles.java | 2 +-
.../fluss/server/log/remote/LogTieringTask.java | 4 +-
.../fluss/server/metadata/ServerSchemaCache.java | 74 ++++++++++++----------
.../server/metadata/TabletServerMetadataCache.java | 10 +--
.../org/apache/fluss/server/replica/Replica.java | 1 +
.../server/metadata/ServerSchemaCacheTest.java | 46 ++++++--------
.../fluss/server/replica/ReplicaManagerTest.java | 16 +++--
.../apache/fluss/server/replica/ReplicaTest.java | 6 +-
.../server/testutils/FlussClusterExtension.java | 12 ++--
12 files changed, 89 insertions(+), 88 deletions(-)
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java b/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java
index b08ad67f8..0ec42a090 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/FetchParams.java
@@ -67,7 +67,7 @@ public final class FetchParams {
private final int minFetchBytes;
private final long maxWaitMs;
- // TODO:add more params like epoch etc.
+ // TODO: add more params like epoch etc.
public FetchParams(int replicaId, int maxFetchBytes) {
this(replicaId, true, maxFetchBytes, DEFAULT_MIN_FETCH_BYTES, DEFAULT_MAX_WAIT_MS);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LocalLog.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LocalLog.java
index 086e23b0e..130b3855d 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/LocalLog.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LocalLog.java
@@ -405,8 +405,6 @@ public final class LocalLog {
(maxOffsetMetadata.getSegmentBaseOffset() == segment.getBaseOffset())
? maxOffsetMetadata.getRelativePositionInSegment()
: segment.getSizeInBytes();
-
- // segment需要查询
fetchDataInfo =
segment.read(readOffset, maxLength, maxPosition, minOneMessage, projection);
if (fetchDataInfo == null) {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/FsRemoteLogOutputStream.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/FsRemoteLogOutputStream.java
index 05233e0ff..835a94ff6 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/FsRemoteLogOutputStream.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/FsRemoteLogOutputStream.java
@@ -203,7 +203,7 @@ public class FsRemoteLogOutputStream extends FSDataOutputStream {
Exception latestException = null;
for (int attempt = 0; attempt < 10; attempt++) {
try {
- // todo: mayadd entropy injection?
+ // todo: may add entropy injection?
this.outStream = fs.create(remoteLogFilePath, FileSystem.WriteMode.NO_OVERWRITE);
return;
} catch (Exception e) {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogSegmentFiles.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogSegmentFiles.java
index 86a6bbf9e..73aee521e 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogSegmentFiles.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogSegmentFiles.java
@@ -40,7 +40,7 @@ public class LogSegmentFiles {
private final Path offsetIndex;
private final Path timeIndex;
private final @Nullable Path writerIdIndex;
- // TODOadd leader epoch index after introduce leader epoch.
+ // TODO add leader epoch index after introduce leader epoch.
public LogSegmentFiles(
Path logSegment, Path offsetIndex, Path timeIndex, @Nullable Path writerIdIndex) {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java
index 0b425b8fc..cdde3842d 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/LogTieringTask.java
@@ -227,7 +227,7 @@ public class LogTieringTask implements Runnable {
}
/**
- * Copy the given log segments to remote andadd the successfully copied segment to the {@code
+ * Copy the given log segments to remote and add the successfully copied segment to the {@code
* copiedSegments} parameter.
*
* @return the end offset of the last segment copied to remote.
@@ -347,7 +347,7 @@ public class LogTieringTask implements Runnable {
// the commit failed, it means the commit snapshot is invalid or register zk
// failed, we will revert this commit and delete the remote log manifest
// file.
- // TODO:add the fail reason in the future.
+ // TODO: add the fail reason in the future.
LOG.error(
"Commit remote log manifest failed for table bucket {}. We will delete the"
+ " written remote log manifest file",
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerSchemaCache.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerSchemaCache.java
index 0f98b092d..374e20501 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerSchemaCache.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerSchemaCache.java
@@ -29,6 +29,8 @@ import org.apache.fluss.shaded.guava32.com.google.common.cache.CacheBuilder;
import org.apache.fluss.shaded.guava32.com.google.common.util.concurrent.UncheckedExecutionException;
import org.apache.fluss.utils.MapUtils;
+import javax.annotation.concurrent.ThreadSafe;
+
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
@@ -41,74 +43,77 @@ import java.util.concurrent.ExecutionException;
* 1. latest schema of each subscribed table, updated by UpdateMetadata request. 2. history schemas
* of each table, updated by lookup from tablet server.
*/
+@ThreadSafe
public class ServerSchemaCache {
private final MetadataManager metadataManager;
- private final Map<TablePath, SchemaInfo> latestSchemaByTablePath;
- private final Map<TablePath, Integer> subscriberCounters;
+ private final Map<Long, SchemaInfo> latestSchemaByTableId;
+ // table_id -> subscriber count
+ private final Map<Long, Integer> subscriberCounters;
private final Cache<TableSchemaKey, Schema> schemaCache;
public ServerSchemaCache(MetadataManager metadataManager) {
this.metadataManager = metadataManager;
// thread safe is guaranteed by subscriberCounters.
this.subscriberCounters = MapUtils.newConcurrentHashMap();
- this.latestSchemaByTablePath = MapUtils.newConcurrentHashMap();
+ this.latestSchemaByTableId = MapUtils.newConcurrentHashMap();
this.schemaCache = CacheBuilder.newBuilder().maximumSize(100).build();
}
public SchemaGetter subscribeWithInitialSchema(
- TablePath tablePath, int initialSchemaId, Schema initialSchema) {
+ long tableId, TablePath tablePath, int initialSchemaId, Schema initialSchema) {
subscriberCounters.compute(
- tablePath,
+ tableId,
(key, oldValue) -> {
- updateSchema(tablePath, initialSchemaId, initialSchema);
+ updateSchema(tableId, initialSchemaId, initialSchema);
if (oldValue == null) {
return 1;
} else {
return oldValue + 1;
}
});
- return new SchemaMetadataSubscriberImpl(tablePath);
+ return new SchemaMetadataSubscriberImpl(tableId, tablePath);
}
- public void updateLatestSchema(TablePath tablePath, short schemaId, Schema schema) {
+ public void updateLatestSchema(long tableId, short schemaId, Schema schema) {
// only update if tablePath is subscribed.
subscriberCounters.computeIfPresent(
- tablePath,
+ tableId,
(key, oldValue) -> {
- updateSchema(tablePath, schemaId, schema);
+ updateSchema(tableId, schemaId, schema);
return oldValue;
});
}
- private void unsubscribe(TablePath tablePath) {
+ private void unsubscribe(long tableId) {
subscriberCounters.compute(
- tablePath,
+ tableId,
(key, oldValue) -> {
if (oldValue != null && oldValue > 1) {
return oldValue - 1;
}
- latestSchemaByTablePath.remove(tablePath);
+ latestSchemaByTableId.remove(tableId);
return null;
});
}
- private Schema getFlussSchema(TablePath tablePath, int schemaId) throws ExecutionException {
+ private Schema getFlussSchema(long tableId, TablePath tablePath, int schemaId)
+ throws ExecutionException {
return schemaCache.get(
- new TableSchemaKey(tablePath, schemaId),
+ new TableSchemaKey(tableId, schemaId),
() -> {
- SchemaInfo schemaInfo = latestSchemaByTablePath.get(tablePath);
+ SchemaInfo schemaInfo = latestSchemaByTableId.get(tableId);
if (schemaInfo != null && schemaInfo.getSchemaId() == schemaId) {
- return latestSchemaByTablePath.get(tablePath).getSchema();
+ return latestSchemaByTableId.get(tableId).getSchema();
} else {
return metadataManager.getSchemaById(tablePath, schemaId).getSchema();
}
});
}
- private void updateSchema(TablePath tablePath, int schemaId, Schema schema) {
- latestSchemaByTablePath.compute(
- tablePath,
+ private void updateSchema(long tableId, int schemaId, Schema schema) {
+ latestSchemaByTableId.compute(
+ tableId,
(key, oldValue) -> {
if (oldValue == null || oldValue.getSchemaId() < schemaId) {
return new SchemaInfo(schema, schemaId);
@@ -117,25 +122,25 @@ public class ServerSchemaCache {
}
});
- schemaCache.put(new TableSchemaKey(tablePath, schemaId), schema);
+ schemaCache.put(new TableSchemaKey(tableId, schemaId), schema);
}
@VisibleForTesting
- public Map<TablePath, SchemaInfo> getLatestSchemaByTablePath() {
- return latestSchemaByTablePath;
+ public Map<Long, SchemaInfo> getLatestSchemaByTableId() {
+ return latestSchemaByTableId;
}
@VisibleForTesting
- Map<TablePath, Integer> getSubscriberCounters() {
+ Map<Long, Integer> getSubscriberCounters() {
return subscriberCounters;
}
private static class TableSchemaKey {
- private final TablePath tablePath;
+ private final long tableId;
private final int schemaId;
- TableSchemaKey(TablePath tablePath, int schemaId) {
- this.tablePath = tablePath;
+ TableSchemaKey(long tableId, int schemaId) {
+ this.tableId = tableId;
this.schemaId = schemaId;
}
@@ -145,34 +150,35 @@ public class ServerSchemaCache {
return false;
}
TableSchemaKey that = (TableSchemaKey) o;
- return schemaId == that.schemaId && Objects.equals(tablePath, that.tablePath);
+ return schemaId == that.schemaId && Objects.equals(tableId, that.tableId);
}
@Override
public int hashCode() {
- return Objects.hash(tablePath, schemaId);
+ return Objects.hash(tableId, schemaId);
}
}
private class SchemaMetadataSubscriberImpl implements SchemaGetter {
+ private final long tableId;
private final TablePath tablePath;
private volatile boolean released;
- public SchemaMetadataSubscriberImpl(TablePath tablePath) {
+ public SchemaMetadataSubscriberImpl(long tableId, TablePath tablePath) {
+ this.tableId = tableId;
this.tablePath = tablePath;
this.released = false;
}
@Override
public SchemaInfo getLatestSchemaInfo() {
- return latestSchemaByTablePath.get(tablePath);
+ return latestSchemaByTableId.get(tableId);
}
@Override
public Schema getSchema(int schemaId) {
- // todo: support with get table.
try {
- return getFlussSchema(tablePath, schemaId);
+ return getFlussSchema(tableId, tablePath, schemaId);
} catch (ExecutionException | UncheckedExecutionException executionException) {
Throwable cause = executionException.getCause();
if (cause instanceof SchemaNotExistException) {
@@ -192,7 +198,7 @@ public class ServerSchemaCache {
@Override
public void release() {
if (!released) {
- unsubscribe(tablePath);
+ unsubscribe(tableId);
released = true;
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
index 9a73975ce..666a74aae 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
@@ -141,14 +141,14 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
}
public SchemaGetter subscribeWithInitialSchema(
- TablePath tablePath, int initialSchemaId, Schema initialSchema) {
+ TablePath tablePath, long tableId, int initialSchemaId, Schema initialSchema) {
return serverSchemaCache.subscribeWithInitialSchema(
- tablePath, initialSchemaId, initialSchema);
+ tableId, tablePath, initialSchemaId, initialSchema);
}
- public void updateLatestSchema(TablePath tablePath, SchemaInfo schemaInfo) {
+ public void updateLatestSchema(long tableId, SchemaInfo schemaInfo) {
serverSchemaCache.updateLatestSchema(
- tablePath, (short) schemaInfo.getSchemaId(), schemaInfo.getSchema());
+ tableId, (short) schemaInfo.getSchemaId(), schemaInfo.getSchema());
}
public Optional<PartitionMetadata> getPartitionMetadata(PhysicalTablePath partitionPath) {
@@ -207,7 +207,7 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
// todo: apply schema id and schema info if needs
int schemaId = tableInfo.getSchemaId();
Schema schema = tableInfo.getSchema();
- serverSchemaCache.updateLatestSchema(tablePath, (short) schemaId, schema);
+ serverSchemaCache.updateLatestSchema(tableId, (short) schemaId, schema);
if (tableId == DELETED_TABLE_ID) {
Long removedTableId = tableIdByPath.remove(tablePath);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
index ad2dcb24d..c1933bd5c 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
@@ -237,6 +237,7 @@ public final class Replica {
this.schemaGetter =
metadataCache.subscribeWithInitialSchema(
physicalPath.getTablePath(),
+ tableInfo.getTableId(),
tableInfo.getSchemaId(),
tableInfo.getSchema());
this.tableConfig = tableInfo.getTableConfig();
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/metadata/ServerSchemaCacheTest.java b/fluss-server/src/test/java/org/apache/fluss/server/metadata/ServerSchemaCacheTest.java
index f72277ba8..c42c7b296 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/metadata/ServerSchemaCacheTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/metadata/ServerSchemaCacheTest.java
@@ -48,36 +48,33 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
public class ServerSchemaCacheTest {
@Test
- void testPublishAndsubscribeSchemaChange() {
+ void testPublishAndSubscribeSchemaChange() {
ServerSchemaCache manager =
new ServerSchemaCache(new TestingMetadataManager(Collections.emptyList()));
SchemaGetter subscriber1 =
manager.subscribeWithInitialSchema(
- new TablePath("test_tb", "test_table_1"), (short) 1, DATA1_SCHEMA);
+ 1L, new TablePath("test_tb", "test_table_1"), (short) 1, DATA1_SCHEMA);
SchemaGetter subscriber2 =
manager.subscribeWithInitialSchema(
- new TablePath("test_tb", "test_table_2"), (short) 1, DATA1_SCHEMA);
+ 2L, new TablePath("test_tb", "test_table_2"), (short) 1, DATA1_SCHEMA);
assertThat(subscriber1.getLatestSchemaInfo()).isEqualTo(new SchemaInfo(DATA1_SCHEMA, 1));
assertThat(subscriber1.getLatestSchemaInfo()).isEqualTo(new SchemaInfo(DATA1_SCHEMA, 1));
- manager.updateLatestSchema(
- new TablePath("test_tb", "test_table_2"), (short) 2, DATA2_SCHEMA);
+ manager.updateLatestSchema(2L, (short) 2, DATA2_SCHEMA);
assertThat(subscriber1.getLatestSchemaInfo()).isNotNull();
assertThat(subscriber1.getLatestSchemaInfo()).isEqualTo(new SchemaInfo(DATA1_SCHEMA, 1));
assertThat(subscriber2.getLatestSchemaInfo()).isNotNull();
assertThat(subscriber2.getLatestSchemaInfo()).isEqualTo(new SchemaInfo(DATA2_SCHEMA, 2));
- manager.updateLatestSchema(
- new TablePath("test_tb", "test_table_1"), (short) 2, DATA2_SCHEMA);
+ manager.updateLatestSchema(1L, (short) 2, DATA2_SCHEMA);
assertThat(subscriber1.getLatestSchemaInfo()).isNotNull();
assertThat(subscriber1.getLatestSchemaInfo()).isEqualTo(new SchemaInfo(DATA2_SCHEMA, 2));
assertThat(subscriber2.getLatestSchemaInfo()).isNotNull();
assertThat(subscriber2.getLatestSchemaInfo()).isEqualTo(new SchemaInfo(DATA2_SCHEMA, 2));
// one more redundant schema change.
- manager.updateLatestSchema(
- new TablePath("test_tb", "test_table_1"), (short) 1, DATA2_SCHEMA);
+ manager.updateLatestSchema(1L, (short) 1, DATA2_SCHEMA);
assertThat(subscriber1.getLatestSchemaInfo()).isNotNull();
assertThat(subscriber1.getLatestSchemaInfo()).isEqualTo(new SchemaInfo(DATA2_SCHEMA, 2));
assertThat(subscriber2.getLatestSchemaInfo()).isNotNull();
@@ -98,7 +95,8 @@ public class ServerSchemaCacheTest {
new ServerSchemaCache(
new TestingMetadataManager(Arrays.asList(DATA1_TABLE_INFO, tableInfo2)));
SchemaGetter schemaGetter =
- manager.subscribeWithInitialSchema(DATA1_TABLE_PATH, (short) 2, DATA2_SCHEMA);
+ manager.subscribeWithInitialSchema(
+ DATA1_TABLE_ID, DATA1_TABLE_PATH, (short) 2, DATA2_SCHEMA);
assertThat(schemaGetter.getSchema(2)).isEqualTo(DATA2_SCHEMA);
assertThat(schemaGetter.getSchema(1)).isEqualTo(DATA1_SCHEMA);
assertThatThrownBy(() -> schemaGetter.getSchema(3))
@@ -111,39 +109,35 @@ public class ServerSchemaCacheTest {
new ServerSchemaCache(new TestingMetadataManager(Collections.emptyList()));
SchemaGetter subscriber1 =
manager.subscribeWithInitialSchema(
- new TablePath("test_tb", "test_table_1"), (short) 1, DATA1_SCHEMA);
+ 1L, new TablePath("test_tb", "test_table_1"), (short) 1, DATA1_SCHEMA);
SchemaGetter subscriber2 =
manager.subscribeWithInitialSchema(
- new TablePath("test_tb", "test_table_1"), (short) 1, DATA1_SCHEMA);
+ 1L, new TablePath("test_tb", "test_table_1"), (short) 1, DATA1_SCHEMA);
SchemaGetter subscriber3 =
manager.subscribeWithInitialSchema(
- new TablePath("test_tb", "test_table_2"), (short) 1, DATA1_SCHEMA);
- assertThat(manager.getLatestSchemaByTablePath()).hasSize(2);
+ 2L, new TablePath("test_tb", "test_table_2"), (short) 1, DATA1_SCHEMA);
+ assertThat(manager.getLatestSchemaByTableId()).hasSize(2);
assertThat(manager.getSubscriberCounters()).hasSize(2);
- assertThat(manager.getSubscriberCounters().get(new TablePath("test_tb", "test_table_1")))
- .isEqualTo(2);
+ assertThat(manager.getSubscriberCounters().get(1L)).isEqualTo(2);
subscriber1.release();
- assertThat(manager.getLatestSchemaByTablePath()).hasSize(2);
+ assertThat(manager.getLatestSchemaByTableId()).hasSize(2);
assertThat(manager.getSubscriberCounters()).hasSize(2);
- assertThat(manager.getSubscriberCounters().get(new TablePath("test_tb", "test_table_1")))
- .isEqualTo(1);
+ assertThat(manager.getSubscriberCounters().get(1L)).isEqualTo(1);
// one more redundant unsubscribe.
subscriber1.release();
- assertThat(manager.getLatestSchemaByTablePath()).hasSize(2);
+ assertThat(manager.getLatestSchemaByTableId()).hasSize(2);
assertThat(manager.getSubscriberCounters()).hasSize(2);
- assertThat(manager.getSubscriberCounters().get(new TablePath("test_tb", "test_table_1")))
- .isEqualTo(1);
+ assertThat(manager.getSubscriberCounters().get(1L)).isEqualTo(1);
subscriber2.release();
- assertThat(manager.getLatestSchemaByTablePath()).hasSize(1);
+ assertThat(manager.getLatestSchemaByTableId()).hasSize(1);
assertThat(manager.getSubscriberCounters()).hasSize(1);
- assertThat(manager.getSubscriberCounters().get(new TablePath("test_tb", "test_table_1")))
- .isNull();
+ assertThat(manager.getSubscriberCounters().get(1L)).isNull();
subscriber3.release();
- assertThat(manager.getLatestSchemaByTablePath()).isEmpty();
+ assertThat(manager.getLatestSchemaByTableId()).isEmpty();
assertThat(manager.getSubscriberCounters()).isEmpty();
subscriber3.release();
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
index 92c24f871..a669af2c1 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
@@ -193,7 +193,7 @@ class ReplicaManagerTest extends ReplicaTestBase {
void testFetchLog() throws Exception {
SchemaGetter schemaGetter =
serverMetadataCache.subscribeWithInitialSchema(
- DATA1_TABLE_PATH, DEFAULT_SCHEMA_ID, DATA1_SCHEMA);
+ DATA1_TABLE_PATH, DATA1_TABLE_ID, DEFAULT_SCHEMA_ID, DATA1_SCHEMA);
TableBucket tb = new TableBucket(DATA1_TABLE_ID, 1);
makeLogTableAsLeader(tb.getBucket());
@@ -309,7 +309,7 @@ class ReplicaManagerTest extends ReplicaTestBase {
void testFetchLogWithMaxBytesLimit() throws Exception {
SchemaGetter schemaGetter =
serverMetadataCache.subscribeWithInitialSchema(
- DATA1_TABLE_PATH, DEFAULT_SCHEMA_ID, DATA1_SCHEMA);
+ DATA1_TABLE_PATH, DATA1_TABLE_ID, DEFAULT_SCHEMA_ID, DATA1_SCHEMA);
TableBucket tb = new TableBucket(DATA1_TABLE_ID, 1);
makeLogTableAsLeader(tb.getBucket());
@@ -432,7 +432,8 @@ class ReplicaManagerTest extends ReplicaTestBase {
assertThat(records1).isNotNull();
assertThat(records2).isNotNull();
SchemaGetter schemaGetter =
- serverMetadataCache.subscribeWithInitialSchema(DATA1_TABLE_PATH, 1, DATA1_SCHEMA);
+ serverMetadataCache.subscribeWithInitialSchema(
+ DATA1_TABLE_PATH, DATA1_TABLE_ID, 1, DATA1_SCHEMA);
if (records1.sizeInBytes() == 0) {
assertThat(records2.sizeInBytes() > 0).isTrue();
assertMemoryRecordsEquals(
@@ -498,7 +499,7 @@ class ReplicaManagerTest extends ReplicaTestBase {
makeKvTableAsLeader(DATA1_TABLE_ID_PK, DATA1_TABLE_PATH_PK, tb.getBucket());
SchemaGetter schemaGetter =
serverMetadataCache.subscribeWithInitialSchema(
- DATA1_TABLE_PATH_PK, 1, DATA1_SCHEMA_PK);
+ DATA1_TABLE_PATH_PK, DATA1_TABLE_ID_PK, 1, DATA1_SCHEMA_PK);
// 1. put kv records to kv store.
List<Tuple2<Object[], Object[]>> data1 =
@@ -628,7 +629,7 @@ class ReplicaManagerTest extends ReplicaTestBase {
makeKvTableAsLeader(DATA1_TABLE_ID_PK, DATA1_TABLE_PATH_PK, tb.getBucket());
SchemaGetter schemaGetter =
serverMetadataCache.subscribeWithInitialSchema(
- DATA1_TABLE_PATH_PK, 1, DATA1_SCHEMA_PK);
+ DATA1_TABLE_PATH_PK, DATA1_TABLE_ID_PK, 1, DATA1_SCHEMA_PK);
// put 10 batches delete non-exists key batch to kv store.
CompletableFuture<List<PutKvResultForBucket>> future;
@@ -906,7 +907,8 @@ class ReplicaManagerTest extends ReplicaTestBase {
CompletableFuture<LimitScanResultForBucket> limitFuture = new CompletableFuture<>();
replicaManager.limitScan(tb, 10, limitFuture::complete);
SchemaGetter schemaGetter =
- serverMetadataCache.subscribeWithInitialSchema(DATA1_TABLE_PATH, 1, DATA1_SCHEMA);
+ serverMetadataCache.subscribeWithInitialSchema(
+ DATA1_TABLE_PATH, DATA1_TABLE_ID, 1, DATA1_SCHEMA);
assertMemoryRecordsEquals(
DATA1_ROW_TYPE,
schemaGetter,
@@ -1463,7 +1465,7 @@ class ReplicaManagerTest extends ReplicaTestBase {
FetchLogResultForBucket r = r1.getValue();
SchemaGetter schemaGetter =
serverMetadataCache.subscribeWithInitialSchema(
- DATA1_TABLE_PATH, 1, DATA1_SCHEMA);
+ DATA1_TABLE_PATH, DATA1_TABLE_ID, 1, DATA1_SCHEMA);
assertLogRecordsEqualsWithRowKind(
DEFAULT_SCHEMA_ID,
DATA1_ROW_TYPE,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
index fcb4ec88c..a18d76442 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
@@ -153,7 +153,7 @@ final class ReplicaTest extends ReplicaTestBase {
// schema evolution.
serverMetadataCache.updateLatestSchema(
- DATA1_TABLE_PATH, new SchemaInfo(DATA2_SCHEMA, (short) 2));
+ DATA1_TABLE_ID, new SchemaInfo(DATA2_SCHEMA, (short) 2));
mr =
createRecordsWithoutBaseLogOffset(
DATA2_ROW_TYPE,
@@ -402,7 +402,7 @@ final class ReplicaTest extends ReplicaTestBase {
currentOffset += 2;
short newSchemaId = 2;
serverMetadataCache.updateLatestSchema(
- DATA1_TABLE_PATH_PK, new SchemaInfo(DATA2_SCHEMA, newSchemaId));
+ DATA1_TABLE_ID, new SchemaInfo(DATA2_SCHEMA, newSchemaId));
KvRecordTestUtils.KvRecordBatchFactory kvRecordBatchFactory2 =
KvRecordTestUtils.KvRecordBatchFactory.of(newSchemaId);
KvRecordTestUtils.KvRecordFactory kvRecordFactory2 =
@@ -727,7 +727,7 @@ final class ReplicaTest extends ReplicaTestBase {
// update schema.
zkClient.registerSchema(DATA1_TABLE_PATH_PK, DATA2_SCHEMA, newSchemaId);
serverMetadataCache.updateLatestSchema(
- DATA1_TABLE_PATH_PK, new SchemaInfo(DATA2_SCHEMA, newSchemaId));
+ DATA1_TABLE_ID, new SchemaInfo(DATA2_SCHEMA, newSchemaId));
// write data with new schema
putRecordsToLeader(
kvReplica,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
index ef1cb92ba..8cd737e51 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
@@ -670,9 +670,9 @@ public final class FlussClusterExtension
() -> {
TableRegistration tableRegistration = zkClient.getTable(tablePath).get();
int bucketCount = tableRegistration.bucketCount;
+ long tableId = tableRegistration.tableId;
for (int bucketId = 0; bucketId < bucketCount; bucketId++) {
- TableBucket tableBucket =
- new TableBucket(tableRegistration.tableId, bucketId);
+ TableBucket tableBucket = new TableBucket(tableId, bucketId);
Optional<LeaderAndIsr> leaderAndIsrOpt =
zkClient.getLeaderAndIsr(tableBucket);
assertThat(leaderAndIsrOpt).isPresent();
@@ -680,10 +680,10 @@ public final class FlussClusterExtension
TabletServer tabletServer = getTabletServerById(leader);
ServerSchemaCache serverSchemaCache =
tabletServer.getMetadataCache().getSchemaMetadataManager();
- Map<TablePath, SchemaInfo> latestSchemaByTablePath =
- serverSchemaCache.getLatestSchemaByTablePath();
- assertThat(latestSchemaByTablePath).containsKey(tablePath);
- assertThat(latestSchemaByTablePath.get(tablePath).getSchemaId())
+ Map<Long, SchemaInfo> latestSchemaByTablePath =
+ serverSchemaCache.getLatestSchemaByTableId();
+ assertThat(latestSchemaByTablePath).containsKey(tableId);
+ assertThat(latestSchemaByTablePath.get(tableId).getSchemaId())
.isEqualTo(schemaId);
}
});
(fluss) 03/03: WIP
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch ci-add-column
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 1e3ecf289ba8a10a18a76dcb97e590897f5e56d9
Author: Jark Wu <ja...@apache.org>
AuthorDate: Mon Dec 1 20:32:22 2025 +0800
WIP
---
.../java/org/apache/fluss/record/BinaryValue.java | 9 +--
.../org/apache/fluss/row/encode/ValueDecoder.java | 12 ----
.../org/apache/fluss/row/encode/ValueEncoder.java | 24 -------
.../java/org/apache/fluss/row/PaddingRowTest.java | 78 ++++++++++++++++++++++
4 files changed, 80 insertions(+), 43 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/BinaryValue.java b/fluss-common/src/main/java/org/apache/fluss/record/BinaryValue.java
index d02bff81b..13906347d 100644
--- a/fluss-common/src/main/java/org/apache/fluss/record/BinaryValue.java
+++ b/fluss-common/src/main/java/org/apache/fluss/record/BinaryValue.java
@@ -18,15 +18,13 @@
package org.apache.fluss.record;
import org.apache.fluss.row.BinaryRow;
-import org.apache.fluss.utils.UnsafeUtils;
+import org.apache.fluss.row.encode.ValueEncoder;
import java.util.Objects;
/** A value of key-value pair that contains schema id and binary row. */
public class BinaryValue {
- public static final int SCHEMA_ID_LENGTH = 2;
-
public final short schemaId;
public final BinaryRow row;
@@ -40,10 +38,7 @@ public class BinaryValue {
* be expected persisted to kv store.
*/
public byte[] encodeValue() {
- byte[] values = new byte[SCHEMA_ID_LENGTH + row.getSizeInBytes()];
- UnsafeUtils.putShort(values, 0, schemaId);
- row.copyTo(values, SCHEMA_ID_LENGTH);
- return values;
+ return ValueEncoder.encodeValue(schemaId, row);
}
@Override
diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java
index c7ac792e4..794e6ad6b 100644
--- a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java
+++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java
@@ -37,7 +37,6 @@ import static org.apache.fluss.utils.MapUtils.newConcurrentHashMap;
*/
public class ValueDecoder {
- // TODO: reuse?
private final Map<Short, RowDecoder> rowDecoders;
private final SchemaGetter schemaGetter;
private final KvFormat kvFormat;
@@ -68,15 +67,4 @@ public class ValueDecoder {
memorySegment, SCHEMA_ID_LENGTH, valueBytes.length - SCHEMA_ID_LENGTH);
return new BinaryValue(schemaId, row);
}
-
- /** The schema id and {@link BinaryRow} stored as the value of kv store. */
- public static class Value {
- public final short schemaId;
- public final BinaryRow row;
-
- private Value(short schemaId, BinaryRow row) {
- this.schemaId = schemaId;
- this.row = row;
- }
- }
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java
index 74eb10094..2d842ab17 100644
--- a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java
+++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java
@@ -17,10 +17,7 @@
package org.apache.fluss.row.encode;
-import org.apache.fluss.metadata.KvFormat;
import org.apache.fluss.row.BinaryRow;
-import org.apache.fluss.row.InternalRow;
-import org.apache.fluss.types.RowType;
import org.apache.fluss.utils.UnsafeUtils;
/** An encoder to encode {@link BinaryRow} with a schema id as value to be stored in kv store. */
@@ -41,25 +38,4 @@ public class ValueEncoder {
row.copyTo(values, SCHEMA_ID_LENGTH);
return values;
}
-
- // TODO: ???
- public static byte[] encodeRow(
- short schemaId, KvFormat kvFormat, RowType currentRowType, InternalRow row)
- throws Exception {
- if (row instanceof BinaryRow) {
- return encodeValue(schemaId, (BinaryRow) row);
- }
- // todo: reuse the encoder here
- try (RowEncoder rowEncoder = RowEncoder.create(kvFormat, currentRowType)) {
- rowEncoder.startNewRow();
- for (int i = 0; i < currentRowType.getFieldCount(); i++) {
- rowEncoder.encodeField(
- i,
- InternalRow.createFieldGetter(currentRowType.getTypeAt(i), i)
- .getFieldOrNull(row));
- }
- BinaryRow binaryRow = rowEncoder.finishRow();
- return encodeValue(schemaId, binaryRow);
- }
- }
}
diff --git a/fluss-common/src/test/java/org/apache/fluss/row/PaddingRowTest.java b/fluss-common/src/test/java/org/apache/fluss/row/PaddingRowTest.java
new file mode 100644
index 000000000..392204b87
--- /dev/null
+++ b/fluss-common/src/test/java/org/apache/fluss/row/PaddingRowTest.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.row;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class PaddingRowTest {
+
+ @Test
+ void testPaddingRowIsNullAt() {
+ // mock an InternalRow with only 2 columns
+ InternalRow row = GenericRow.of(1, null);
+ PaddingRow paddingRow = new PaddingRow(4).replaceRow(row);
+
+ assertThat(paddingRow.getFieldCount()).isEqualTo(4);
+ assertThat(paddingRow.isNullAt(0)).isFalse();
+ assertThat(paddingRow.isNullAt(1)).isTrue();
+ assertThat(paddingRow.isNullAt(2)).isTrue(); // exceed original row, should be null
+ assertThat(paddingRow.isNullAt(3)).isTrue();
+ }
+
+ @Test
+ void testPaddingRowDelegatesGetters() {
+ InternalRow row =
+ GenericRow.of(
+ true,
+ (byte) 2,
+ (short) 3,
+ 4,
+ 5L,
+ 6.5f,
+ 7.5,
+ new byte[] {1, 2, 3},
+ new byte[] {4, 5},
+ BinaryString.fromString("abc"),
+ TimestampLtz.fromEpochMillis(1000));
+ PaddingRow paddingRow = new PaddingRow(9).replaceRow(row);
+
+ assertThat(paddingRow.getBoolean(0)).isTrue();
+ assertThat(paddingRow.getByte(1)).isEqualTo((byte) 2);
+ assertThat(paddingRow.getShort(2)).isEqualTo((short) 3);
+ assertThat(paddingRow.getInt(3)).isEqualTo(4);
+ assertThat(paddingRow.getLong(4)).isEqualTo(5L);
+ assertThat(paddingRow.getFloat(5)).isEqualTo(6.5f);
+ assertThat(paddingRow.getDouble(6)).isEqualTo(7.5);
+ assertThat(paddingRow.getBinary(7, 3)).isEqualTo(new byte[] {1, 2, 3});
+ assertThat(paddingRow.getBytes(8)).isEqualTo(new byte[] {4, 5});
+ assertThat(paddingRow.getString(9)).isEqualTo(BinaryString.fromString("abc"));
+ assertThat(paddingRow.getTimestampLtz(10, 3)).isEqualTo(TimestampLtz.fromEpochMillis(1000));
+ }
+
+ @Test
+ void testReplaceRowInPlace() {
+ InternalRow row1 = GenericRow.of(1);
+ InternalRow row2 = GenericRow.of(2);
+ PaddingRow paddingRow = new PaddingRow(1).replaceRow(row1);
+ assertThat(paddingRow.getInt(0)).isEqualTo(1);
+ paddingRow.replaceRow(row2);
+ assertThat(paddingRow.getInt(0)).isEqualTo(2);
+ }
+}
(fluss) 02/03: WIP
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch ci-add-column
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 639c492037d6a91d214b1e145ed8574113ced652
Author: Jark Wu <ja...@apache.org>
AuthorDate: Mon Dec 1 20:23:30 2025 +0800
WIP
---
.../org/apache/fluss/server/RpcServiceBase.java | 2 +-
.../fluss/server/coordinator/MetadataManager.java | 4 +-
.../server/metadata/TabletServerMetadataCache.java | 3 +-
.../fluss/server/replica/AdjustIsrManager.java | 2 +-
.../org/apache/fluss/server/replica/Replica.java | 4 +-
.../replica/fetcher/ReplicaFetcherManager.java | 2 +-
.../replica/fetcher/ReplicaFetcherThread.java | 2 +-
.../fluss/server/utils/RpcGatewayManager.java | 2 +-
.../fluss/server/utils/ServerRpcMessageUtils.java | 2 +-
.../org/apache/fluss/server/utils/timer/Timer.java | 2 +-
.../fluss/server/utils/timer/TimingWheel.java | 4 +-
.../apache/fluss/server/zk/ZooKeeperClient.java | 50 +++++++++++++++++-----
.../server/authorizer/DefaultAuthorizerTest.java | 4 +-
.../CompletedSnapshotStoreManagerTest.java | 8 ++--
.../coordinator/CoordinatorChannelManagerTest.java | 6 +--
.../fluss/server/coordinator/TableManagerTest.java | 2 -
.../server/kv/prewrite/KvPreWriteBufferTest.java | 4 +-
.../kv/snapshot/CompletedSnapshotStoreTest.java | 2 +-
.../kv/snapshot/KvTabletSnapshotTargetTest.java | 2 +-
.../ZooKeeperCompletedSnapshotStoreTest.java | 4 +-
.../apache/fluss/server/log/LogManagerTest.java | 2 +-
.../org/apache/fluss/server/log/LogTabletTest.java | 4 +-
.../server/log/remote/RemoteLogTabletTest.java | 4 +-
.../server/metadata/MetadataUpdateITCase.java | 2 +-
.../fluss/server/replica/AdjustIsrITCase.java | 2 +-
.../fluss/server/replica/AdjustIsrManagerTest.java | 2 +-
.../replica/HighWatermarkPersistenceTest.java | 2 +-
.../replica/RemoveOfflineReplicaFromIsrITCase.java | 2 +-
.../apache/fluss/server/replica/ReplicaTest.java | 4 +-
.../fluss/server/replica/ReplicaTestBase.java | 4 +-
.../fluss/server/tablet/TabletServiceITCase.java | 4 +-
.../server/testutils/FlussClusterExtension.java | 4 +-
fluss-test-coverage/pom.xml | 4 +-
33 files changed, 88 insertions(+), 63 deletions(-)
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
index ededf510f..82f3d5ec2 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
@@ -384,7 +384,7 @@ public abstract class RpcServiceBase extends RpcGatewayService implements AdminR
@Override
public CompletableFuture<GetFileSystemSecurityTokenResponse> getFileSystemSecurityToken(
GetFileSystemSecurityTokenRequest request) {
- // TODO:add ACL for per-table in https://github.com/apache/fluss/issues/752
+ // TODO: add ACL for per-table in https://github.com/apache/fluss/issues/752
try {
// In order to avoid repeatedly obtaining security token, cache it for a while.
long currentTimeMs = System.currentTimeMillis();
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
index bf2cfc9fb..3d63c8ba1 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
@@ -576,14 +576,14 @@ public class MetadataManager {
zookeeperClient.getTables(tablePaths);
// currently, we don't support schema evolution, so all schemas are version 1
Map<TablePath, SchemaInfo> tablePath2SchemaInfos =
- zookeeperClient.getV1Schemas(tablePaths);
+ zookeeperClient.getLatestSchemas(tablePaths);
for (TablePath tablePath : tablePaths) {
if (!tablePath2TableRegistrations.containsKey(tablePath)) {
throw new TableNotExistException("Table '" + tablePath + "' does not exist.");
}
if (!tablePath2SchemaInfos.containsKey(tablePath)) {
throw new SchemaNotExistException(
- "Schema for '" + tablePath + "' with schema_id=1 does not exist.");
+ "Schema for '" + tablePath + "' does not exist.");
}
TableRegistration tableReg = tablePath2TableRegistrations.get(tablePath);
SchemaInfo schemaInfo = tablePath2SchemaInfos.get(tablePath);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
index 666a74aae..f4ecff6f0 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
@@ -70,7 +70,6 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
private final ServerSchemaCache serverSchemaCache;
- // todo: replace this in test with schemaMetadataManager.
public TabletServerMetadataCache(MetadataManager metadataManager) {
this.serverMetadataSnapshot = ServerMetadataSnapshot.empty();
this.metadataManager = metadataManager;
@@ -436,7 +435,7 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
}
@VisibleForTesting
- public ServerSchemaCache getSchemaMetadataManager() {
+ public ServerSchemaCache getServerSchemaCache() {
return serverSchemaCache;
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/AdjustIsrManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/AdjustIsrManager.java
index 251d670b6..d3b5d63c9 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/AdjustIsrManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/AdjustIsrManager.java
@@ -77,7 +77,7 @@ public class AdjustIsrManager {
public CompletableFuture<LeaderAndIsr> submit(
TableBucket tableBucket, LeaderAndIsr leaderAndIsr) {
- // TODOadd coordinatorEpoch.
+ // TODO add coordinatorEpoch.
CompletableFuture<LeaderAndIsr> future = new CompletableFuture<>();
AdjustIsrItem adjustIsrItem = new AdjustIsrItem(tableBucket, leaderAndIsr, future);
boolean enqueued = unsentAdjustIsrMap.putIfAbsent(tableBucket, adjustIsrItem) == null;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
index c1933bd5c..d8c88cbd7 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
@@ -1518,7 +1518,7 @@ public final class Replica {
List<Integer> isrToSend = new ArrayList<>(isrState.isr());
isrToSend.add(newInSyncReplicaId);
- // TODOadd server epoch to isr.
+ // TODO add server epoch to isr.
LeaderAndIsr newLeaderAndIsr =
new LeaderAndIsr(
@@ -1540,7 +1540,7 @@ public final class Replica {
// erroneously advance the HW if the `AdjustIsr` were to fail. Hence, the "maximal ISR"
// for `PendingShrinkIsr` is the current ISR.
- // TODOadd server epoch to isr.
+ // TODO add server epoch to isr.
LeaderAndIsr newLeaderAndIsr =
new LeaderAndIsr(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManager.java
index 06e94e9ae..9e4479b3e 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManager.java
@@ -225,7 +225,7 @@ public class ReplicaFetcherManager {
fetcherThread.getLeader().leaderServerId(),
initialFetchStatusMap);
} catch (InterruptedException e) {
- LOG.error("Interrupted whileadd buckets to fetcher threads.", e);
+ LOG.error("Interrupted while add buckets to fetcher threads.", e);
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThread.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThread.java
index 76195a0df..398bfb6ff 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThread.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThread.java
@@ -80,7 +80,7 @@ final class ReplicaFetcherThread extends ShutdownableThread {
private final LeaderEndpoint leader;
private final int fetchBackOffMs;
- // manuallyadd timout logic in here, todo remove this timeout logic if
+ // manually add timout logic in here, todo remove this timeout logic if
// we support global request timeout in #279
private final int timeoutSeconds = 30;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/RpcGatewayManager.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/RpcGatewayManager.java
index 062eec73d..c89fca6f7 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/RpcGatewayManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/RpcGatewayManager.java
@@ -65,7 +65,7 @@ public class RpcGatewayManager<T extends RpcGateway> implements AutoCloseable {
}
/**
- * Add a server to the manager. It'll create a new gateway for the server andadd it to the
+ * Add a server to the manager. It'll create a new gateway for the server and add it to the
* manager. If the server has already existed, it'll remove the already existing server before
* adding the new one.
*/
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
index 165fd23e9..815771166 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
@@ -339,7 +339,7 @@ public class ServerRpcMessageUtils {
.collect(Collectors.toList());
}
- private static @Nullable TableChange.ColumnPosition toColumnPosition(int columnPositionType) {
+ private static TableChange.ColumnPosition toColumnPosition(int columnPositionType) {
ColumnPositionType opType = ColumnPositionType.from(columnPositionType);
switch (opType) {
case LAST:
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/timer/Timer.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/timer/Timer.java
index 270d334ee..ea6aead7b 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/timer/Timer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/timer/Timer.java
@@ -35,7 +35,7 @@ public interface Timer {
* Add a new task to this executor. It will be executed after the task's delay (beginning from
* the time of submission).
*
- * @param timerTask the task toadd
+ * @param timerTask the task to add
*/
void add(TimerTask timerTask);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/timer/TimingWheel.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/timer/TimingWheel.java
index 934777c93..1dab9f229 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/timer/TimingWheel.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/timer/TimingWheel.java
@@ -108,8 +108,8 @@ import java.util.concurrent.atomic.AtomicInteger;
* while priority queue based timers takes O(log N) for both insert and delete where N is the number
* of items in the queue.
*
- * <p>This class is not thread-safe. There should not be anyadd calls while advanceClock is
- * executing. It is caller's responsibility to enforce it. Simultaneousadd calls are thread-safe.
+ * <p>This class is not thread-safe. There should not be any add calls while advanceClock is
+ * executing. It is caller's responsibility to enforce it. Simultaneous add calls are thread-safe.
*/
@NotThreadSafe
final class TimingWheel {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
index e27ffd9c3..e80006e60 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
@@ -536,18 +536,46 @@ public class ZooKeeperClient implements AutoCloseable {
"tables registration");
}
- /** Get the v1 schema for given tables in ZK. */
- public Map<TablePath, SchemaInfo> getV1Schemas(Collection<TablePath> tablePaths)
+ /** Get the latest schema for given tables in ZK. */
+ public Map<TablePath, SchemaInfo> getLatestSchemas(Collection<TablePath> tablePaths)
throws Exception {
- Map<String, TablePath> path2TablePathMap =
- tablePaths.stream()
- .collect(toMap(p -> SchemaZNode.path(p, DEFAULT_SCHEMA_ID), path -> path));
+ Map<String, TablePath> schemaChildren2TablePathMap =
+ tablePaths.stream().collect(toMap(SchemasZNode::path, path -> path));
+ List<ZkGetChildrenResponse> childrenResponses =
+ getChildrenInBackground(schemaChildren2TablePathMap.keySet());
+ // get the schema ids for each table
+ Map<TablePath, List<String>> schemaIdsForTables =
+ processGetChildrenResponses(
+ childrenResponses,
+ response -> schemaChildren2TablePathMap.get(response.getPath()),
+ "schema children for tables");
+
+ // get the schema info for each latest schema id
+ Map<TablePath, Integer> latestSchemaIdMap = new HashMap<>();
+ Map<String, TablePath> path2TablePathMap = new HashMap<>();
+ schemaIdsForTables.forEach(
+ (tp, schemaIds) -> {
+ int latestSchemaId =
+ schemaIds.stream().map(Integer::parseInt).reduce(Math::max).orElse(0);
+ latestSchemaIdMap.put(tp, latestSchemaId);
+ path2TablePathMap.put(SchemaZNode.path(tp, latestSchemaId), tp);
+ });
+
List<ZkGetDataResponse> responses = getDataInBackground(path2TablePathMap.keySet());
- return processGetDataResponses(
- responses,
- resp -> path2TablePathMap.get(resp.getPath()),
- b -> new SchemaInfo(SchemaZNode.decode(b), DEFAULT_SCHEMA_ID),
- "schema");
+ Map<TablePath, Schema> schemasForTables =
+ processGetDataResponses(
+ responses,
+ resp -> path2TablePathMap.get(resp.getPath()),
+ SchemaZNode::decode,
+ "schema");
+
+ Map<TablePath, SchemaInfo> result = new HashMap<>();
+ schemasForTables.forEach(
+ (tp, schema) -> {
+ int schemaId = latestSchemaIdMap.get(tp);
+ result.put(tp, new SchemaInfo(schema, schemaId));
+ });
+ return result;
}
/** Update the table in ZK. */
@@ -810,7 +838,7 @@ public class ZooKeeperClient implements AutoCloseable {
/** Register schema to ZK metadata and return the schema id. */
public int registerSchema(TablePath tablePath, Schema schema) throws Exception {
- return registerSchema(tablePath, schema, getCurrentSchemaId(tablePath) + 1);
+ return registerSchema(tablePath, schema, DEFAULT_SCHEMA_ID);
}
public int registerSchema(TablePath tablePath, Schema schema, int schemaId) throws Exception {
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java
index abc575fd5..af69caffe 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java
@@ -487,7 +487,7 @@ public class DefaultAuthorizerTest {
Set<AccessControlEntry> expectedAcls = new HashSet<>();
for (int i = 0; i < 50; i++) {
- // each task represent that addColumn acl to different user on same resource.
+ // each task represent that add acl to different user on same resource.
final AccessControlEntry accessControlEntry =
createAclEntry(String.valueOf(i), "host-1", READ);
int finalI = i;
@@ -550,7 +550,7 @@ public class DefaultAuthorizerTest {
// generate 50 concurrent acl operation tasks.
List<Runnable> concurrentTasks = new ArrayList<>();
for (int i = 0; i < 50; i++) {
- // each task represent that addColumn acl to same user on same resource.
+ // each task represent that add acl to same user on same resource.
Runnable runnable =
() -> {
addAcls(
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java
index 4460cfc26..b8fec59ca 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java
@@ -99,13 +99,13 @@ class CompletedSnapshotStoreManagerTest {
CompletedSnapshotStoreManager completedSnapshotStoreManager =
createCompletedSnapshotStoreManager(maxNumberOfSnapshotsToRetain);
- // addColumn snapshots for a series of buckets
+ // add snapshots for a series of buckets
Set<TableBucket> tableBuckets = createTableBuckets(2, 3);
int snapshotNum = 3;
Map<TableBucket, CompletedSnapshot> tableBucketLatestCompletedSnapshots = new HashMap<>();
for (TableBucket tableBucket : tableBuckets) {
- // addColumn some snapshots
+ // add some snapshots
for (int snapshot = 0; snapshot < snapshotNum; snapshot++) {
CompletedSnapshot completedSnapshot =
KvTestUtils.mockCompletedSnapshot(tempDir, tableBucket, snapshot);
@@ -137,7 +137,7 @@ class CompletedSnapshotStoreManagerTest {
assertThat(completedSnapshot)
.isEqualTo(tableBucketLatestCompletedSnapshots.get(tableBucket));
- // addColumn a new snapshot
+ // add a new snapshot
long snapshotId = completedSnapshot.getSnapshotID() + 1;
completedSnapshot = KvTestUtils.mockCompletedSnapshot(tempDir, tableBucket, snapshotId);
addCompletedSnapshot(completedSnapshotStoreManager, completedSnapshot);
@@ -169,7 +169,7 @@ class CompletedSnapshotStoreManagerTest {
Set<TableBucket> tableBuckets = createTableBuckets(1, 2);
int snapshotNum = 3;
for (TableBucket tableBucket : tableBuckets) {
- // addColumn some snapshots
+ // add some snapshots
for (int snapshot = 0; snapshot < snapshotNum; snapshot++) {
CompletedSnapshot completedSnapshot =
KvTestUtils.mockCompletedSnapshot(tempDir, tableBucket, snapshot);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorChannelManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorChannelManagerTest.java
index 581edafcc..bd4467c0a 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorChannelManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorChannelManagerTest.java
@@ -69,14 +69,14 @@ class CoordinatorChannelManagerTest {
// now, shouldn't send as we already remove the tablet server
checkSendRequest(coordinatorChannelManager, server0.id(), false);
- // test addColumn tablet server
- // before addColumn, shouldn't send
+ // test add tablet server
+ // before add, shouldn't send
ServerNode server1 = tabletServersNode.get(1);
checkSendRequest(coordinatorChannelManager, server1.id(), false);
coordinatorChannelManager.addTabletServer(server1);
- // after addColumn the tablet server, should send
+ // after add the tablet server, should send
// try to send message
checkSendRequest(coordinatorChannelManager, server1.id(), true);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
index b8b348bdb..3c417efe8 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
@@ -211,8 +211,6 @@ class TableManagerTest {
assertThat(coordinatorContext.getAllReplicasForTable(tableId)).isEmpty();
}
- void testSchemaChange() throws Exception {}
-
@Test
void testResumeDeletionAfterRestart() throws Exception {
// first, create a table
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java
index 74d145bda..b39ec39ac 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java
@@ -163,13 +163,13 @@ class KvPreWriteBufferTest {
}
assertThat(getValue(buffer, "key6")).isNull();
- // addColumn delete records.
+ // add delete records.
elementCount = 5;
bufferDelete(buffer, "key4", elementCount++);
bufferDelete(buffer, "key3", elementCount++);
assertThat(getValue(buffer, "key3")).isNull();
- // addColumn update records
+ // add update records
bufferPut(buffer, "key2", "value2-1", elementCount++);
bufferPut(buffer, "key1", "value1-1", elementCount++);
assertThat(getValue(buffer, "key1")).isEqualTo("value1-1");
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java
index 87fb23259..8d45628d8 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java
@@ -161,7 +161,7 @@ class CompletedSnapshotStoreTest {
final CompletedSnapshot ckp = getSnapshot(ckpId);
assertThatThrownBy(() -> completedSnapshotStore.add(ckp))
- .as("We should get an exception when addColumn snapshot to failed..")
+ .as("We should get an exception when add snapshot to failed..")
.hasMessageContaining(errMsg)
.isInstanceOf(FlussException.class);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTargetTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTargetTest.java
index 2451e7ecc..9f9ea078d 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTargetTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTargetTest.java
@@ -247,7 +247,7 @@ class KvTabletSnapshotTargetTest {
final String errMsg = "Add to snapshot handle failed.";
AtomicBoolean shouldFail = new AtomicBoolean(true);
// we use a store will fail when the variable shouldFail is true
- // addColumn the snapshot to store will fail
+ // add the snapshot to store will fail
CompletedSnapshotHandleStore completedSnapshotHandleStore =
TestCompletedSnapshotHandleStore.newBuilder()
.setAddFunction(
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotStoreTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotStoreTest.java
index 5b890acd6..c6e90ff63 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotStoreTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotStoreTest.java
@@ -94,8 +94,8 @@ class ZooKeeperCompletedSnapshotStoreTest {
}
/**
- * Tests that the snapshot does not exist in the store when we fail to addColumn it into the
- * store (i.e., there exists an exception thrown by the method).
+ * Tests that the snapshot does not exist in the store when we fail to add it into the store
+ * (i.e., there exists an exception thrown by the method).
*/
@Test
void testAddSnapshotWithFailedRemove(@TempDir Path tmpDir) throws Exception {
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java
index f4e536b86..2bbdd2a1c 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java
@@ -79,7 +79,7 @@ final class LogManagerTest extends LogTestBase {
private TableBucket tableBucket2;
private LogManager logManager;
- // TODOadd more tests refer to kafka's LogManagerTest.
+ // TODO add more tests refer to kafka's LogManagerTest.
@BeforeAll
static void baseBeforeAll() {
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
index 7af086823..7b88a53a5 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
@@ -73,7 +73,7 @@ final class LogTabletTest extends LogTestBase {
private FlussScheduler scheduler;
private File logDir;
- // TODOadd more tests refer to kafka's UnifiedLogTest.
+ // TODO add more tests refer to kafka's UnifiedLogTest.
@BeforeEach
public void setup() throws Exception {
@@ -374,7 +374,7 @@ final class LogTabletTest extends LogTestBase {
log.updateHighWatermark(log.localLogEndOffset());
- // TODOadd delete log segment logic.
+ // TODO add delete log segment logic.
}
@Test
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java
index 5bedd921e..0d571e1ea 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTabletTest.java
@@ -94,7 +94,7 @@ class RemoteLogTabletTest extends RemoteLogTestBase {
assertThat(remoteLogTablet.getRemoteLogStartOffset()).isEqualTo(0);
assertThat(remoteLogTablet.getRemoteLogEndOffset()).isEqualTo(OptionalLong.of(30));
- // delete first remote log segment and addColumn another one remote log segments.
+ // delete first remote log segment and add another one remote log segments.
remoteLogTablet.addAndDeleteLogSegments(
Collections.singletonList(remoteLogSegmentList.get(3)),
Collections.singletonList(remoteLogSegmentList.get(0)));
@@ -102,7 +102,7 @@ class RemoteLogTabletTest extends RemoteLogTestBase {
assertThat(remoteLogTablet.getRemoteLogStartOffset()).isEqualTo(10);
assertThat(remoteLogTablet.getRemoteLogEndOffset()).isEqualTo(OptionalLong.of(40));
- // delete all exist and append one. we will first addColumn then delete.
+ // delete all exist and append one. we will first add then delete.
remoteLogTablet.addAndDeleteLogSegments(
Collections.singletonList(remoteLogSegmentList.get(4)), remoteLogSegmentList);
assertThat(remoteLogTablet.getIdToRemoteLogSegmentMap()).hasSize(0);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/metadata/MetadataUpdateITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/metadata/MetadataUpdateITCase.java
index 35c36d30c..b6fdd9e93 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/metadata/MetadataUpdateITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/metadata/MetadataUpdateITCase.java
@@ -157,7 +157,7 @@ class MetadataUpdateITCase {
expectedTablePathById,
Collections.emptyMap()));
- // addColumn back tablet server2
+ // add back tablet server2
FLUSS_CLUSTER_EXTENSION.startTabletServer(2);
retry(
Duration.ofMinutes(1),
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrITCase.java
index bb2138881..933abd231 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrITCase.java
@@ -140,7 +140,7 @@ public class AdjustIsrITCase {
isr.add(stopFollower);
FLUSS_CLUSTER_EXTENSION.notifyLeaderAndIsr(
stopFollower, DATA1_TABLE_PATH, tb, newLeaderAndIsr, isr);
- // retry until the stop follower addColumn back to ISR.
+ // retry until the stop follower add back to ISR.
retry(
Duration.ofMinutes(1),
() ->
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrManagerTest.java
index 75fb9e248..3f8623569 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrManagerTest.java
@@ -32,7 +32,7 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link AdjustIsrManager}. */
class AdjustIsrManagerTest {
- // TODO addColumn more test refer to kafka AlterPartitionManagerTest, See: FLUSS-56278513
+ // TODO add more test refer to kafka AlterPartitionManagerTest, See: FLUSS-56278513
@Test
void testSubmitShrinkIsr() throws Exception {
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/HighWatermarkPersistenceTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/HighWatermarkPersistenceTest.java
index b089efd30..b093787f0 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/HighWatermarkPersistenceTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/HighWatermarkPersistenceTest.java
@@ -92,7 +92,7 @@ final class HighWatermarkPersistenceTest extends ReplicaTestBase {
assertThat(highWatermark0).isEqualTo(10L);
assertThat(replica0.getLogTablet().getHighWatermark()).isEqualTo(10L);
- // addColumn another replica and set highWatermark.
+ // add another replica and set highWatermark.
TableBucket tableBucket1 = new TableBucket(DATA2_TABLE_ID, 0);
replicaManager.becomeLeaderOrFollower(
INITIAL_COORDINATOR_EPOCH,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/RemoveOfflineReplicaFromIsrITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/RemoveOfflineReplicaFromIsrITCase.java
index a7c65acda..9e13b1688 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/RemoveOfflineReplicaFromIsrITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/RemoveOfflineReplicaFromIsrITCase.java
@@ -130,7 +130,7 @@ class RemoveOfflineReplicaFromIsrITCase {
FLUSS_CLUSTER_EXTENSION.startTabletServer(follower);
isr.add(follower);
- // make sure the stopped follower can addColumn back to isr after restart
+ // make sure the stopped follower can add back to isr after restart
retry(
Duration.ofMinutes(1),
() ->
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
index a18d76442..c8e26ea7b 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
@@ -95,8 +95,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link Replica}. */
final class ReplicaTest extends ReplicaTestBase {
- // TODO addColumn more tests refer to kafka's PartitionTest.
- // TODO addColumn more tests to cover partition table
+ // TODO add more tests refer to kafka's PartitionTest.
+ // TODO add more tests to cover partition table
@Test
void testMakeLeader() throws Exception {
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
index c438da529..80d998a4d 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
@@ -342,7 +342,7 @@ public class ReplicaTestBase {
}
// TODO this is only for single tablet server unit test.
- // TODO addColumn more test cases for partition table which make leader by this method.
+ // TODO add more test cases for partition table which make leader by this method.
protected void makeLogTableAsLeader(int bucketId) {
makeLogTableAsLeader(new TableBucket(DATA1_TABLE_ID, bucketId), false);
}
@@ -375,7 +375,7 @@ public class ReplicaTestBase {
}
// TODO this is only for single tablet server unit test.
- // TODO addColumn more test cases for partition table which make leader by this method.
+ // TODO add more test cases for partition table which make leader by this method.
protected void makeKvTableAsLeader(long tableId, TablePath tablePath, int bucketId) {
makeKvTableAsLeader(
new TableBucket(tableId, bucketId), tablePath, INITIAL_LEADER_EPOCH, false);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java
index 8b0b75c35..01cf2701a 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java
@@ -177,7 +177,7 @@ public class TabletServiceITCase {
}
@Test
- @Disabled("TODO: addColumn back in https://github.com/apache/fluss/issues/771")
+ @Disabled("TODO: add back in https://github.com/apache/fluss/issues/771")
void testProduceLogResponseReturnInOrder() throws Exception {
long tableId =
createTable(
@@ -334,7 +334,7 @@ public class TabletServiceITCase {
}
@Test
- @Disabled("TODO: addColumn back in https://github.com/apache/fluss/issues/777")
+ @Disabled("TODO: add back in https://github.com/apache/fluss/issues/777")
void testFetchLogWithMinFetchSizeAndTimeout() throws Exception {
long tableId =
createTable(FLUSS_CLUSTER_EXTENSION, DATA1_TABLE_PATH, DATA1_TABLE_DESCRIPTOR);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
index 8cd737e51..a1ee83fa6 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
@@ -286,7 +286,7 @@ public final class FlussClusterExtension
}
private void startTabletServers() throws Exception {
- // addColumn tablet server to make generate assignment for table possible
+ // add tablet server to make generate assignment for table possible
for (int i = 0; i < initialNumOfTabletServers; i++) {
startTabletServer(i);
}
@@ -679,7 +679,7 @@ public final class FlussClusterExtension
int leader = leaderAndIsrOpt.get().leader();
TabletServer tabletServer = getTabletServerById(leader);
ServerSchemaCache serverSchemaCache =
- tabletServer.getMetadataCache().getSchemaMetadataManager();
+ tabletServer.getMetadataCache().getServerSchemaCache();
Map<Long, SchemaInfo> latestSchemaByTablePath =
serverSchemaCache.getLatestSchemaByTableId();
assertThat(latestSchemaByTablePath).containsKey(tableId);
diff --git a/fluss-test-coverage/pom.xml b/fluss-test-coverage/pom.xml
index ca167c2d9..0c6f54cd5 100644
--- a/fluss-test-coverage/pom.xml
+++ b/fluss-test-coverage/pom.xml
@@ -164,7 +164,7 @@
<excludes>
<exclude>fluss-test-coverage/**</exclude>
<exclude>fluss-test-utils/**</exclude>
- <!-- exclude adapter classes to avoid Jacoco error: "Can'tadd different class with same name" -->
+ <!-- exclude adapter classes to avoid Jacoco error: "Can't add different class with same name" -->
<exclude>fluss-flink/**/target/classes/org/apache/fluss/flink/adapter/**</exclude>
</excludes>
</resource>
@@ -206,7 +206,7 @@
<excludes>
<exclude>fluss-test-coverage/**</exclude>
<exclude>fluss-test-utils/**</exclude>
- <!-- exclude adapter classes to avoid Jacoco error: "Can'tadd different class with same name" -->
+ <!-- exclude adapter classes to avoid Jacoco error: "Can't add different class with same name" -->
<exclude>fluss-flink/**/target/classes/org/apache/fluss/flink/adapter/**</exclude>
</excludes>
</resource>