[server] Add rebalance id to trace rebalance task #2286
Conversation
swuferhong
force-pushed
the
rebalance-id
branch
2 times, most recently
from
fbd1e6a to
1782631
Compare
There was a problem hiding this comment.
Pull request overview
This PR adds a rebalance ID field to trace rebalance tasks throughout the system. The implementation includes adding the rebalanceId field to RebalancePlan, updating serialization/deserialization logic, implementing comprehensive rebalance management functionality, and adding corresponding client APIs and Flink procedures.
Key Changes
- Added
rebalanceIdfield toRebalancePlanfor task tracing - Implemented
RebalanceManagerfor coordinating rebalance operations - Refactored replica leader election from enum-based to class-based strategy pattern
- Added Flink procedures for rebalance operations (rebalance, cancel, list progress)
Reviewed changes
Copilot reviewed 72 out of 72 changed files in this pull request and generated 10 comments.
Show a summary per file
| File | Description |
|---|---|
| RebalancePlan.java | Added rebalanceId field with getter and updated equals/hashCode |
| RebalancePlanJsonSerde.java | Updated JSON serialization to include rebalance_id field |
| ZooKeeperClient.java | Added updatePartitionAssignment and deleteRebalancePlan methods |
| RebalanceManager.java | New manager class for coordinating rebalance operations |
| CoordinatorEventProcessor.java | Added rebalance task execution logic and replica reassignment |
| ReplicaLeaderElection.java | Refactored from enum to class-based strategy pattern |
| Multiple test files | Added comprehensive test coverage for rebalance functionality |
| Flink procedures | Added procedures for rebalance, cancel, and list progress operations |
| Config files | Added TABLE_GENERATE_UNBALANCE_TABLE_ASSIGNMENT config option |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| public void registerRebalance( | ||
| String rebalanceId, Map<TableBucket, RebalancePlanForBucket> rebalancePlan) { | ||
| checkNotClosed(); | ||
| registerTime = System.currentTimeMillis(); |
There was a problem hiding this comment.
This field access (publicly accessible via this expression) is not protected by any monitor, but the class is annotated as @threadsafe.
| /** Test for {@link GoalOptimizer}. */ | ||
| public class GoalOptimizerTest { | ||
|
|
||
| private SortedSet<ServerModel> servers; |
There was a problem hiding this comment.
The contents of this container are never accessed.
| } catch (Exception e) { | ||
| LOG.error( | ||
| "Failed to get rebalance plan from zookeeper, it will be treated as no" | ||
| + "rebalance tasks.", |
There was a problem hiding this comment.
This string appears to be missing a space after 'no'.
| + "rebalance tasks.", | |
| + " rebalance tasks.", |
| rebalancePlan = | ||
| rebalanceManager.generateRebalancePlan(rebalanceEvent.getGoalsByPriority()); | ||
| } catch (Exception e) { | ||
| throw new RebalanceFailureException("Failed to generate rebalance plan.", e); |
| throw new RebalanceFailureException( | ||
| "Rebalance task already exists. Please wait for it to finish or cancel it first."); |
| throw new RebalanceFailureException( | ||
| "Rebalance task already exists. Please wait for it to finish or cancel it first."); |
| } catch (Exception e) { | ||
| LOG.error("Error when register rebalance plan to zookeeper.", e); | ||
| throw new RebalanceFailureException( | ||
| "Error when register rebalance plan to zookeeper.", e); |
| throw new RebalanceFailureException( | ||
| String.format( | ||
| "[%s] All alive tabletServers are excluded from replica moves.", | ||
| name())); |
| LOG.debug("Starting Optimizing for goal {}", name()); | ||
| // Initialize pre-optimized stats. | ||
| ClusterModelStats statsBeforeOptimization = clusterModel.getClusterStats(); | ||
| LOG.trace("[PRE - {}] {}", name(), statsBeforeOptimization); |
There was a problem hiding this comment.
Default toString(): ClusterModelStats inherits toString() from Object, and so is not suitable for printing.
| } | ||
|
|
||
| ClusterModelStats statsAfterOptimization = clusterModel.getClusterStats(); | ||
| LOG.trace("[POST - {}] {}", name(), statsAfterOptimization); |
There was a problem hiding this comment.
Default toString(): ClusterModelStats inherits toString() from Object, and so is not suitable for printing.
swuferhong
force-pushed
the
rebalance-id
branch
from
6396d5b to
f2133e8
Compare
swuferhong
force-pushed
the
rebalance-id
branch
from
f2133e8 to
654239f
Compare
1 participant
Purpose
[server] Add rebalance id to trace rebalance task
Linked issue: close #2285
Brief change log
Tests
API and Format
Documentation