[Issue #1215] Implement NodeService, Retina Node Heartbeat, and Consistent Hashing by AntiO2 · Pull Request #1216 · pixelsdb/pixels

AntiO2 · 2025-11-24T09:46:30Z

No description provided.

Copilot

Pull request overview

This pull request implements a NodeService with consistent hashing for managing Retina nodes, adds heartbeat functionality for Retina nodes, and integrates these components into the daemon architecture.

Introduces gRPC-based NodeService with dynamic consistent hashing for Retina node discovery and routing
Extends HeartbeatWorker to support Retina nodes in addition to Worker nodes
Adds NodeServer to the coordinator daemon for centralized node management

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 21 comments.

Show a summary per file

File	Description
proto/node.proto	Defines gRPC service and messages for node management operations including GetRetinaList and GetRetinaByBucket
pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/node/NodeServiceImpl.java	Implements NodeService with consistent hashing ring, etcd-based node discovery, and dynamic node add/remove
pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/node/NodeServer.java	Creates gRPC server wrapper for NodeService
pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/heartbeat/HeartbeatWorker.java	Extends heartbeat support to Retina nodes using role-based etcd key registration
pixels-daemon/src/main/java/io/pixelsdb/pixels/daemon/DaemonMain.java	Integrates NodeServer into coordinator startup and uses NodeRole enum for type-safe role handling
pixels-common/src/main/java/io/pixelsdb/pixels/common/node/NodeService.java	Provides client-side gRPC stub for communicating with NodeService
pixels-common/src/main/java/io/pixelsdb/pixels/common/utils/Constants.java	Adds HEARTBEAT_RETINA_LITERAL constant for etcd key prefixing
pixels-common/src/main/java/io/pixelsdb/pixels/common/error/ErrorCode.java	Defines error codes for node service operations
pixels-common/src/main/resources/pixels.properties	Adds configuration properties for node server port, virtual nodes, and bucket count
pixels-common/src/test/java/io/pixelsdb/pixels/common/node/TestNodeService.java	Provides basic tests for NodeService client operations
pixels-daemon/src/test/java/io/pixelsdb/pixels/daemon/heartbeat/TestHeartbeatWorker.java	Updates test to use NodeRole parameter for HeartbeatWorker

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-11-24T10:46:28Z

+        ConfigFactory config = ConfigFactory.Instance();
+        virtualNode = Integer.parseInt(config.getProperty("node.virtual.num"));
+        // bucketNum is the total number of hash points
+        bucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));


Missing input validation for configuration properties. The constructor should validate that virtualNode and bucketNum are positive integers, and that virtualNode is less than or equal to bucketNum to ensure proper consistent hashing. Invalid configuration values could cause runtime errors or incorrect behavior.

Suggested change

bucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));

bucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));

// Validate configuration values

if (virtualNode <= 0) {

throw new IllegalArgumentException("Configuration error: node.virtual.num must be a positive integer, got " + virtualNode);

}

if (bucketNum <= 0) {

throw new IllegalArgumentException("Configuration error: node.bucket.num must be a positive integer, got " + bucketNum);

}

if (virtualNode > bucketNum) {

throw new IllegalArgumentException("Configuration error: node.virtual.num (" + virtualNode + ") must be less than or equal to node.bucket.num (" + bucketNum + ")");

}

Copilot · 2025-11-24T10:46:28Z

+ * gRPC implementation of NodeService with dynamic consistent hashing.
+ * Supports adding/removing Retina nodes at runtime.
+ * Implementation is modified to use bucketNum as the number of virtual nodes,
+ * and assumes the client provides a pre-hashed key (bucket ID).


The class-level documentation on lines 43-44 mentions "assumes the client provides a pre-hashed key (bucket ID)" but this is misleading. Looking at the getRetinaByBucket method, the client provides a bucket ID directly (0 to bucketNum-1), not a pre-hashed key. Consider updating the documentation to clarify that clients should provide a bucket ID in the range [0, bucketNum-1].

Suggested change

* and assumes the client provides a pre-hashed key (bucket ID).

* and assumes the client provides a bucket ID in the range [0, bucketNum-1].

Copilot · 2025-11-24T10:46:32Z

+node.server.port=18891
+node.server.host=localhost
+# virtual nodes num
+node.virtual.num=16


[nitpick] Add a comment to explain what node.bucket.num represents. This should clarify that it's the total number of hash buckets in the consistent hash ring.

Suggested change

node.virtual.num=16

node.virtual.num=16

# the total number of hash buckets in the consistent hash ring

Copilot · 2025-11-24T10:46:33Z

                int metadataServerPort = Integer.parseInt(config.getProperty("metadata.server.port"));
                int transServerPort = Integer.parseInt(config.getProperty("trans.server.port"));
                int queryScheduleServerPort = Integer.parseInt(config.getProperty("query.schedule.server.port"));
                int scalingMetricsServerPort = Integer.parseInt(config.getProperty("scaling.metrics.server.port"));
+                int nodeServerPort = Integer.parseInt(config.getProperty("node.server.port"));


Potential uncaught 'java.lang.NumberFormatException'.

Suggested change

int metadataServerPort = Integer.parseInt(config.getProperty("metadata.server.port"));

int transServerPort = Integer.parseInt(config.getProperty("trans.server.port"));

int queryScheduleServerPort = Integer.parseInt(config.getProperty("query.schedule.server.port"));

int scalingMetricsServerPort = Integer.parseInt(config.getProperty("scaling.metrics.server.port"));

int nodeServerPort = Integer.parseInt(config.getProperty("node.server.port"));

int metadataServerPort;

int transServerPort;

int queryScheduleServerPort;

int scalingMetricsServerPort;

int nodeServerPort;

try {

metadataServerPort = Integer.parseInt(config.getProperty("metadata.server.port"));

} catch (NumberFormatException e) {

log.error("Invalid value for metadata.server.port: " + config.getProperty("metadata.server.port"), e);

throw e;

}

try {

transServerPort = Integer.parseInt(config.getProperty("trans.server.port"));

} catch (NumberFormatException e) {

log.error("Invalid value for trans.server.port: " + config.getProperty("trans.server.port"), e);

throw e;

}

try {

queryScheduleServerPort = Integer.parseInt(config.getProperty("query.schedule.server.port"));

} catch (NumberFormatException e) {

log.error("Invalid value for query.schedule.server.port: " + config.getProperty("query.schedule.server.port"), e);

throw e;

}

try {

scalingMetricsServerPort = Integer.parseInt(config.getProperty("scaling.metrics.server.port"));

} catch (NumberFormatException e) {

log.error("Invalid value for scaling.metrics.server.port: " + config.getProperty("scaling.metrics.server.port"), e);

throw e;

}

try {

nodeServerPort = Integer.parseInt(config.getProperty("node.server.port"));

} catch (NumberFormatException e) {

log.error("Invalid value for node.server.port: " + config.getProperty("node.server.port"), e);

throw e;

}

Copilot · 2025-11-24T10:46:34Z

+        virtualNode = Integer.parseInt(config.getProperty("node.virtual.num"));
+        // bucketNum is the total number of hash points
+        bucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));


Potential uncaught 'java.lang.NumberFormatException'.

Suggested change

virtualNode = Integer.parseInt(config.getProperty("node.virtual.num"));

// bucketNum is the total number of hash points

bucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));

int tmpVirtualNode;

int tmpBucketNum;

try {

tmpVirtualNode = Integer.parseInt(config.getProperty("node.virtual.num"));

} catch (NumberFormatException e) {

logger.error("Invalid value for 'node.virtual.num': {}", config.getProperty("node.virtual.num"), e);

throw new IllegalArgumentException("Invalid configuration for 'node.virtual.num'", e);

}

// bucketNum is the total number of hash points

try {

tmpBucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));

} catch (NumberFormatException e) {

logger.error("Invalid value for 'node.bucket.num': {}", config.getProperty("node.bucket.num"), e);

throw new IllegalArgumentException("Invalid configuration for 'node.bucket.num'", e);

}

virtualNode = tmpVirtualNode;

bucketNum = tmpBucketNum;

Copilot · 2025-11-24T10:46:34Z

+        virtualNode = Integer.parseInt(config.getProperty("node.virtual.num"));
+        // bucketNum is the total number of hash points
+        bucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));


Potential uncaught 'java.lang.NumberFormatException'.

Suggested change

virtualNode = Integer.parseInt(config.getProperty("node.virtual.num"));

// bucketNum is the total number of hash points

bucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));

int tmpVirtualNode = 1;

try {

tmpVirtualNode = Integer.parseInt(config.getProperty("node.virtual.num"));

} catch (NumberFormatException e) {

logger.error("Invalid value for node.virtual.num in config, using default: 1", e);

}

virtualNode = tmpVirtualNode;

// bucketNum is the total number of hash points

int tmpBucketNum = 1;

try {

tmpBucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));

} catch (NumberFormatException e) {

logger.error("Invalid value for node.bucket.num in config, using default: 1", e);

}

bucketNum = tmpBucketNum;

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

bianhq

LGTM.

Node Service

d888524

AntiO2 requested a review from bianhq November 24, 2025 09:46

gengdy1545 requested a review from Copilot November 24, 2025 10:37

Copilot started reviewing on behalf of gengdy1545 November 24, 2025 10:37 View session

Copilot finished reviewing on behalf of gengdy1545 November 24, 2025 10:41

Copilot AI reviewed Nov 24, 2025

View reviewed changes

Apply suggestions from code review

bbcbbe9

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

bianhq requested changes Nov 24, 2025

View reviewed changes

Change License

a25a079

AntiO2 force-pushed the issue/1215 branch from dd8888c to a25a079 Compare November 24, 2025 14:33

AntiO2 requested a review from bianhq November 24, 2025 14:35

bianhq approved these changes Nov 25, 2025

View reviewed changes

bianhq merged commit d559156 into pixelsdb:master Nov 25, 2025
1 check passed

bianhq assigned AntiO2 Nov 25, 2025

bianhq added the enhancement New feature or request label Nov 25, 2025

bianhq added this to the Real-time CRUD milestone Nov 25, 2025

bianhq linked an issue Nov 25, 2025 that may be closed by this pull request

[pixels-daemon, common, proto] Implement Retina Heartbeat Tracking and consistent hash #1215

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Issue #1215] Implement NodeService, Retina Node Heartbeat, and Consistent Hashing#1216

[Issue #1215] Implement NodeService, Retina Node Heartbeat, and Consistent Hashing#1216
bianhq merged 3 commits intopixelsdb:masterfrom
AntiO2:issue/1215

AntiO2 commented Nov 24, 2025

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI Nov 24, 2025

Uh oh!

Copilot AI Nov 24, 2025

Uh oh!

Copilot AI Nov 24, 2025

Uh oh!

Uh oh!

Copilot AI Nov 24, 2025

Uh oh!

Copilot AI Nov 24, 2025

Uh oh!

Copilot AI Nov 24, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

bianhq left a comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

-        bucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));
+        bucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));
+        // Validate configuration values
+        if (virtualNode <= 0) {
+            throw new IllegalArgumentException("Configuration error: node.virtual.num must be a positive integer, got " + virtualNode);
+        }
+        if (bucketNum <= 0) {
+            throw new IllegalArgumentException("Configuration error: node.bucket.num must be a positive integer, got " + bucketNum);
+        }
+        if (virtualNode > bucketNum) {
+            throw new IllegalArgumentException("Configuration error: node.virtual.num (" + virtualNode + ") must be less than or equal to node.bucket.num (" + bucketNum + ")");
+        }

	* and assumes the client provides a pre-hashed key (bucket ID).
	* and assumes the client provides a bucket ID in the range [0, bucketNum-1].

	node.virtual.num=16
	node.virtual.num=16
	# the total number of hash buckets in the consistent hash ring

-                int metadataServerPort = Integer.parseInt(config.getProperty("metadata.server.port"));
-                int transServerPort = Integer.parseInt(config.getProperty("trans.server.port"));
-                int queryScheduleServerPort = Integer.parseInt(config.getProperty("query.schedule.server.port"));
-                int scalingMetricsServerPort = Integer.parseInt(config.getProperty("scaling.metrics.server.port"));
-                int nodeServerPort = Integer.parseInt(config.getProperty("node.server.port"));
+                int metadataServerPort;
+                int transServerPort;
+                int queryScheduleServerPort;
+                int scalingMetricsServerPort;
+                int nodeServerPort;
+                try {
+                    metadataServerPort = Integer.parseInt(config.getProperty("metadata.server.port"));
+                } catch (NumberFormatException e) {
+                    log.error("Invalid value for metadata.server.port: " + config.getProperty("metadata.server.port"), e);
+                    throw e;
+                }
+                try {
+                    transServerPort = Integer.parseInt(config.getProperty("trans.server.port"));
+                } catch (NumberFormatException e) {
+                    log.error("Invalid value for trans.server.port: " + config.getProperty("trans.server.port"), e);
+                    throw e;
+                }
+                try {
+                    queryScheduleServerPort = Integer.parseInt(config.getProperty("query.schedule.server.port"));
+                } catch (NumberFormatException e) {
+                    log.error("Invalid value for query.schedule.server.port: " + config.getProperty("query.schedule.server.port"), e);
+                    throw e;
+                }
+                try {
+                    scalingMetricsServerPort = Integer.parseInt(config.getProperty("scaling.metrics.server.port"));
+                } catch (NumberFormatException e) {
+                    log.error("Invalid value for scaling.metrics.server.port: " + config.getProperty("scaling.metrics.server.port"), e);
+                    throw e;
+                }
+                try {
+                    nodeServerPort = Integer.parseInt(config.getProperty("node.server.port"));
+                } catch (NumberFormatException e) {
+                    log.error("Invalid value for node.server.port: " + config.getProperty("node.server.port"), e);
+                    throw e;
+                }

-        virtualNode = Integer.parseInt(config.getProperty("node.virtual.num"));
-        // bucketNum is the total number of hash points
-        bucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));
+        int tmpVirtualNode;
+        int tmpBucketNum;
+        try {
+            tmpVirtualNode = Integer.parseInt(config.getProperty("node.virtual.num"));
+        } catch (NumberFormatException e) {
+            logger.error("Invalid value for 'node.virtual.num': {}", config.getProperty("node.virtual.num"), e);
+            throw new IllegalArgumentException("Invalid configuration for 'node.virtual.num'", e);
+        }
+        // bucketNum is the total number of hash points
+        try {
+            tmpBucketNum = Integer.parseInt(config.getProperty("node.bucket.num"));
+        } catch (NumberFormatException e) {
+            logger.error("Invalid value for 'node.bucket.num': {}", config.getProperty("node.bucket.num"), e);
+            throw new IllegalArgumentException("Invalid configuration for 'node.bucket.num'", e);
+        }
+        virtualNode = tmpVirtualNode;
+        bucketNum = tmpBucketNum;

Conversation

AntiO2 commented Nov 24, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI Nov 24, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 24, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 24, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Copilot AI Nov 24, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 24, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 24, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

bianhq left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants