You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pt...@apache.org on 2025/02/24 11:01:23 UTC
(ignite-3) 01/09: IGNITE-24581 Data Streamer: validate partition assignment size
This is an automated email from the ASF dual-hosted git repository.
ptupitsyn pushed a commit to branch ignite-24581
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 54ef38a03fe8ac770e0d4f91b53fe54c1b30bbd4
Author: Pavel Tupitsyn <pt...@apache.org>
AuthorDate: Fri Feb 21 11:10:39 2025 +0200
IGNITE-24581 Data Streamer: validate partition assignment size
---
.../AbstractClientStreamerPartitionAwarenessProvider.java | 11 ++++++++---
.../org/apache/ignite/internal/client/table/ClientTable.java | 4 +++-
2 files changed, 11 insertions(+), 4 deletions(-)
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/AbstractClientStreamerPartitionAwarenessProvider.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/AbstractClientStreamerPartitionAwarenessProvider.java
index b79616c0efd..f2201597630 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/AbstractClientStreamerPartitionAwarenessProvider.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/AbstractClientStreamerPartitionAwarenessProvider.java
@@ -40,7 +40,11 @@ abstract class AbstractClientStreamerPartitionAwarenessProvider<T> implements St
int partitions0 = partitions;
if (schema0 == null || partitions0 < 0) {
- throw new IllegalStateException("StreamerPartitionAwarenessProvider.refresh() was not called or awaited.");
+ throw new IllegalStateException("StreamerPartitionAwarenessProvider.refreshAsync() was not called or awaited.");
+ }
+
+ if (partitions0 == 0) {
+ throw new IllegalStateException("Partition count is zero.");
}
int hash = colocationHash(schema0, item);
@@ -53,7 +57,7 @@ abstract class AbstractClientStreamerPartitionAwarenessProvider<T> implements St
public CompletableFuture<Void> refreshAsync() {
var schemaFut = tbl.getLatestSchema().thenAccept(schema -> this.schema = schema);
- if (partitions < 0) {
+ if (partitions <= 0) {
partitions = tbl.tryGetPartitionCount();
}
@@ -62,7 +66,8 @@ abstract class AbstractClientStreamerPartitionAwarenessProvider<T> implements St
return schemaFut;
}
- var assignmentFut = tbl.getPartitionAssignment().thenAccept(assignment -> this.partitions = assignment.size());
+ var assignmentFut = tbl.getPartitionAssignment()
+ .thenAccept(assignment -> this.partitions = assignment.size());
return CompletableFuture.allOf(schemaFut, assignmentFut);
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
index 95f27cfd9a7..5869a70f7d7 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
@@ -633,7 +633,9 @@ public class ClientTable implements Table {
},
r -> {
int cnt = r.in().unpackInt();
- assert cnt >= 0 : "Invalid partition count: " + cnt;
+ if (cnt <= 0) {
+ throw new IgniteException(INTERNAL_ERR, "Invalid partition count returned by the server: " + cnt);
+ }
int oldPartitionCount = partitionCount;