1919
2020package org .apache .fluss .client .admin ;
2121
22+ import org .apache .fluss .cluster .rebalance .ServerTag ;
2223import org .apache .fluss .config .AutoPartitionTimeUnit ;
2324import org .apache .fluss .config .ConfigOptions ;
2425import org .apache .fluss .exception .TableNotPartitionedException ;
2526import org .apache .fluss .metadata .PartitionInfo ;
2627import org .apache .fluss .metadata .Schema ;
28+ import org .apache .fluss .metadata .TableBucket ;
2729import org .apache .fluss .metadata .TableDescriptor ;
2830import org .apache .fluss .metadata .TablePath ;
2931import org .apache .fluss .rpc .messages .ListPartitionInfosRequest ;
3335
3436import org .junit .jupiter .api .Test ;
3537
38+ import java .time .Duration ;
39+ import java .util .Collections ;
3640import java .util .List ;
3741import java .util .Map ;
3842
3943import static org .apache .fluss .metadata .DataLakeFormat .PAIMON ;
4044import static org .apache .fluss .record .TestData .DATA1_PARTITIONED_TABLE_DESCRIPTOR ;
4145import static org .apache .fluss .record .TestData .DATA1_TABLE_DESCRIPTOR_PK ;
46+ import static org .apache .fluss .testutils .common .CommonTestUtils .waitUntil ;
4247import static org .apache .fluss .utils .PartitionUtils .HISTORICAL_PARTITION_VALUE ;
4348import static org .assertj .core .api .Assertions .assertThat ;
4449import static org .assertj .core .api .Assertions .assertThatThrownBy ;
@@ -164,6 +169,94 @@ void testListPartitionInfos() throws Exception {
164169 .doesNotContain (HISTORICAL_PARTITION_VALUE );
165170 }
166171
172+ @ Test
173+ void testDescribeTabletServersReportsServerTags () throws Exception {
174+ admin .addServerTag (Collections .singletonList (0 ), ServerTag .PERMANENT_OFFLINE ).get ();
175+ try {
176+ List <TabletServerDescription > servers = admin .describeTabletServers ().get ();
177+ assertThat (getTabletServerDescription (servers , 0 ).getServerTag ())
178+ .contains (ServerTag .PERMANENT_OFFLINE );
179+ assertThat (getTabletServerDescription (servers , 1 ).getServerTag ()).isNotPresent ();
180+ assertThat (getTabletServerDescription (servers , 2 ).getServerTag ()).isNotPresent ();
181+ } finally {
182+ admin .removeServerTag (Collections .singletonList (0 ), ServerTag .PERMANENT_OFFLINE ).get ();
183+ }
184+ assertThat (admin .describeTabletServers ().get ())
185+ .allSatisfy (server -> assertThat (server .getServerTag ()).isNotPresent ());
186+ }
187+
188+ @ Test
189+ void testDescribeTabletServersDuringRollingUpgrade () throws Exception {
190+ TablePath tablePath = TablePath .of ("test_db" , "describe_tablet_servers_table" );
191+ long tableId = createTable (tablePath , DATA1_TABLE_DESCRIPTOR_PK , true );
192+ waitAllReplicasReady (tableId , 3 );
193+
194+ // Phase 1: Cluster is healthy - every live server is reported, hosts replicas of the
195+ // created table (replication factor 3 on 3 servers) and is green. The cluster is shared
196+ // with other tests, so wait until residue from them (e.g. a recovering ISR) has settled
197+ // before taking the snapshot asserted below.
198+ waitUntil (
199+ () ->
200+ admin .describeTabletServers ().get ().stream ()
201+ .allMatch (FlussAdmin2ITCase ::isServerGreen ),
202+ Duration .ofMinutes (1 ),
203+ "All tablet servers should be green before the rolling upgrade starts" );
204+
205+ List <TabletServerDescription > servers = admin .describeTabletServers ().get ();
206+ assertThat (servers ).extracting (TabletServerDescription ::getServerId ).contains (0 , 1 , 2 );
207+ for (TabletServerDescription server : servers ) {
208+ assertThat (server .getNumReplicas ()).isGreaterThan (0 );
209+ assertThat (isServerGreen (server )).isTrue ();
210+ }
211+
212+ // The per-server counters must sum up to the cluster-wide health counters.
213+ ClusterHealth health = admin .getClusterHealth ().get ();
214+ assertThat (servers .stream ().mapToInt (TabletServerDescription ::getNumReplicas ).sum ())
215+ .isEqualTo (health .getNumReplicas ());
216+ assertThat (servers .stream ().mapToInt (TabletServerDescription ::getInSyncReplicas ).sum ())
217+ .isEqualTo (health .getInSyncReplicas ());
218+ assertThat (servers .stream ().mapToInt (TabletServerDescription ::getNumLeaderReplicas ).sum ())
219+ .isEqualTo (health .getNumLeaderReplicas ());
220+ assertThat (
221+ servers .stream ()
222+ .mapToInt (TabletServerDescription ::getActiveLeaderReplicas )
223+ .sum ())
224+ .isEqualTo (health .getActiveLeaderReplicas ());
225+
226+ // Phase 2: Stop one tablet server (simulate server crash during rolling upgrade). It must
227+ // still be reported with its assigned replicas, but no longer green - an operator must
228+ // not treat it as safe to remove.
229+ int stoppedServerId = 0 ;
230+ FLUSS_CLUSTER_EXTENSION .stopTabletServer (stoppedServerId );
231+ FLUSS_CLUSTER_EXTENSION .assertHasTabletServerNumber (2 );
232+
233+ for (int bucket = 0 ; bucket < 3 ; bucket ++) {
234+ TableBucket tb = new TableBucket (tableId , bucket );
235+ FLUSS_CLUSTER_EXTENSION .waitUntilReplicaShrinkFromIsr (tb , stoppedServerId );
236+ }
237+
238+ TabletServerDescription stopped =
239+ getTabletServerDescription (admin .describeTabletServers ().get (), stoppedServerId );
240+ assertThat (stopped .getNumReplicas ()).isGreaterThan (0 );
241+ assertThat (stopped .getInSyncReplicas ()).isLessThan (stopped .getNumReplicas ());
242+
243+ // Phase 3: Restart the server and wait until every server is green again.
244+ FLUSS_CLUSTER_EXTENSION .startTabletServer (stoppedServerId );
245+ FLUSS_CLUSTER_EXTENSION .assertHasTabletServerNumber (3 );
246+
247+ for (int bucket = 0 ; bucket < 3 ; bucket ++) {
248+ TableBucket tb = new TableBucket (tableId , bucket );
249+ FLUSS_CLUSTER_EXTENSION .waitUntilReplicaExpandToIsr (tb , stoppedServerId );
250+ }
251+
252+ waitUntil (
253+ () ->
254+ admin .describeTabletServers ().get ().stream ()
255+ .allMatch (FlussAdmin2ITCase ::isServerGreen ),
256+ Duration .ofMinutes (1 ),
257+ "All tablet servers should become green again after server restart" );
258+ }
259+
167260 private static ListPartitionInfosRequest requestFor (
168261 TablePath tablePath , boolean includeSystemPartitions ) {
169262 ListPartitionInfosRequest request =
@@ -177,4 +270,20 @@ private static ListPartitionInfosRequest requestFor(
177270 }
178271 return request ;
179272 }
273+
274+ private static TabletServerDescription getTabletServerDescription (
275+ List <TabletServerDescription > servers , int serverId ) {
276+ return servers .stream ()
277+ .filter (server -> server .getServerId () == serverId )
278+ .findFirst ()
279+ .orElseThrow (
280+ () ->
281+ new AssertionError (
282+ "no description reported for tablet server " + serverId ));
283+ }
284+
285+ private static boolean isServerGreen (TabletServerDescription server ) {
286+ return server .getInSyncReplicas () == server .getNumReplicas ()
287+ && server .getActiveLeaderReplicas () == server .getNumLeaderReplicas ();
288+ }
180289}
0 commit comments