fix: prevent CPU runaway in HttpNodesStatsAction JSON parsing (#18)

Three parse methods introduced in #14 were missing `parser.nextToken()`
before calling `consumeObject()` or sub-parse methods when the current
token was START_OBJECT. This caused `consumeObject()` to consume tokens
beyond its scope (including sibling fields), corrupting the parser state
and eventually leading to an infinite loop at EOF — spinning all eshttp
threads at 100% CPU each.

Fixes:
- Add `parser.nextToken()` in parseSearchBackpressureStats,
  parseTaskCancellationStats, and parseSearchPipelineStats
- Add null-token (EOF) guards in fromXContent, parseNodes,
  parseNodeStats, and consumeObject to throw IOException instead of
  spinning
- Add START_ARRAY handling in consumeObject
- Remove orphan `new ArrayList<>()` in parseNodeStats

Tests: add 61 unit tests covering token boundary verification, field
ordering, multiple nodes, partial sub-fields, deeply nested structures,
truncated JSON, and concurrent parsing.

Author: marevol

src/main/java/org/codelibs/fesen/client/action/HttpNodesStatsAction.java

@@ -159,6 +159,9 @@ protected NodesStatsResponse fromXContent(final XContentParser parser) throws IO
         ClusterName clusterName = ClusterName.DEFAULT;
         XContentParser.Token token;
         while ((token = parser.currentToken()) != XContentParser.Token.END_OBJECT) {
+            if (token == null) {
+                throw new IOException("Unexpected end of JSON stream while parsing NodesStatsResponse");
+            }
             if (token == XContentParser.Token.FIELD_NAME) {
                 fieldName = parser.currentName();
             } else if (token == XContentParser.Token.START_OBJECT) {
@@ -181,6 +184,9 @@ protected List<NodeStats> parseNodes(final XContentParser parser) throws IOExcep
         String fieldName = null;
         XContentParser.Token token;
         while ((token = parser.currentToken()) != XContentParser.Token.END_OBJECT) {
+            if (token == null) {
+                throw new IOException("Unexpected end of JSON stream while parsing nodes");
+            }
             if (token == XContentParser.Token.FIELD_NAME) {
                 fieldName = parser.currentName();
             } else if (token == XContentParser.Token.START_OBJECT) {
@@ -193,7 +199,6 @@ protected List<NodeStats> parseNodes(final XContentParser parser) throws IOExcep
     }
 
     protected NodeStats parseNodeStats(final XContentParser parser, final String nodeId) throws IOException {
-        new ArrayList<>();
         String fieldName = null;
         String nodeName = "";
         long timestamp = 0;
@@ -230,6 +235,9 @@ protected NodeStats parseNodeStats(final XContentParser parser, final String nod
         XContentParser.Token token;
         TransportAddress transportAddress = new TransportAddress(TransportAddress.META_ADDRESS, 0);
         while ((token = parser.currentToken()) != XContentParser.Token.END_OBJECT) {
+            if (token == null) {
+                throw new IOException("Unexpected end of JSON stream while parsing node stats for " + nodeId);
+            }
             if (token == XContentParser.Token.FIELD_NAME) {
                 fieldName = parser.currentName();
             } else if (token == XContentParser.Token.START_OBJECT) {
@@ -399,6 +407,7 @@ protected SearchBackpressureStats parseSearchBackpressureStats(final XContentPar
             if (token == XContentParser.Token.FIELD_NAME) {
                 fieldName = parser.currentName();
             } else if (token == XContentParser.Token.START_OBJECT) {
+                parser.nextToken();
                 if ("search_task".equals(fieldName) || "search_shard_task".equals(fieldName)) {
                     consumeObject(parser);
                 } else {
@@ -544,6 +553,7 @@ protected TaskCancellationStats parseTaskCancellationStats(final XContentParser
             if (token == XContentParser.Token.FIELD_NAME) {
                 fieldName = parser.currentName();
             } else if (token == XContentParser.Token.START_OBJECT) {
+                parser.nextToken();
                 if ("search_task".equals(fieldName)) {
                     searchTaskCancellationStats = parseSearchTaskCancellationStats(parser);
                 } else if ("search_shard_task".equals(fieldName)) {
@@ -606,6 +616,7 @@ protected SearchPipelineStats parseSearchPipelineStats(final XContentParser pars
             if (token == XContentParser.Token.FIELD_NAME) {
                 fieldName = parser.currentName();
             } else if (token == XContentParser.Token.START_OBJECT) {
+                parser.nextToken();
                 if ("total_request".equals(fieldName)) {
                     totalRequestStats = parseOperationStats(parser);
                 } else if ("total_response".equals(fieldName)) {
@@ -2322,9 +2333,14 @@ protected int[] parseNodeResults(final XContentParser parser) throws IOException
     protected void consumeObject(final XContentParser parser) throws IOException {
         XContentParser.Token token;
         while ((token = parser.currentToken()) != XContentParser.Token.END_OBJECT) {
+            if (token == null) {
+                throw new IOException("Unexpected end of JSON stream while consuming object");
+            }
             if (token == XContentParser.Token.START_OBJECT) {
                 parser.nextToken();
                 consumeObject(parser);
+            } else if (token == XContentParser.Token.START_ARRAY) {
+                parser.skipChildren();
             }
             parser.nextToken();
         }