Skip to content

Commit b2ff369

Browse files
dudanogueiraclaude
andcommitted
fix(tenants): split tenant updates at the server's limit of 100
PUT /schema/{class}/tenants rejects more than 100 tenants with HTTP 422, so updating more than that failed outright -- and with it activate, deactivate and offload, which all delegate to update. The Python and TypeScript clients split the request internally, so the same code written against either of them worked and the Java one did not. update() now sends the tenants in batches of 100, on the sync and async clients alike; the async one chains them so they reach the server one after another rather than all at once. Adding tenants stays a single request: the server validates it with allowOverHundred=true, so only updates are capped. More than one request means a partial update is now possible: if a batch fails, the tenants of the preceding batches stay updated and the error propagates. Python and TypeScript behave the same way; it is documented on both update() methods. MockRestTransport.performRequestAsync returned null, which no chained caller could compose on; it returns a completed future now. Closes #615 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WmY5dAGWCccWDoqkKNC2JU
1 parent c59a3fd commit b2ff369

6 files changed

Lines changed: 215 additions & 4 deletions

File tree

‎src/it/java/io/weaviate/integration/TenantsITest.java‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,8 @@
11
package io.weaviate.integration;
22

3+
import java.util.List;
4+
import java.util.stream.IntStream;
5+
36
import org.assertj.core.api.Assertions;
47
import org.junit.Test;
58

@@ -71,4 +74,35 @@ public void test_tenantLifecycle() throws Exception {
7174
Assertions.assertThat(things.tenants.exists(owen.name()))
7275
.describedAs("%s not exists", owen.name()).isFalse();
7376
}
77+
78+
/**
79+
* The server refuses to update more than 100 tenants at once, so the client
80+
* splits longer lists -- otherwise activate/deactivate of 101+ tenants fails
81+
* with HTTP 422 ("maximum number of tenants allowed to be updated
82+
* simultaneously is 100").
83+
*/
84+
@Test
85+
public void test_updateMoreThanOneHundredTenants() throws Exception {
86+
// Arrange: 250 tenants, comfortably over two batch boundaries.
87+
var nsMany = ns("ManyTenants");
88+
client.collections.create(nsMany, c -> c.multiTenancy(mt -> mt.autoTenantCreation(false)));
89+
var many = client.collections.use(nsMany);
90+
91+
List<String> names = IntStream.rangeClosed(1, 250).mapToObj(i -> "tenant-" + i).toList();
92+
// Creating is not capped by the server and goes out in a single request.
93+
many.tenants.create(names.stream().map(Tenant::active).toList());
94+
Assertions.assertThat(many.tenants.list()).as("created").hasSize(250);
95+
96+
// Act
97+
many.tenants.deactivate(names);
98+
99+
// Assert
100+
eventually(() -> many.tenants.list().stream().allMatch(Tenant::isInactive),
101+
200, 5, "not all tenants were deactivated");
102+
103+
// ...and back, to cover the activate path as well.
104+
many.tenants.activate(names);
105+
eventually(() -> many.tenants.list().stream().allMatch(Tenant::isActive),
106+
200, 5, "not all tenants were activated");
107+
}
74108
}

‎src/main/java/io/weaviate/client6/v1/api/collections/tenants/UpdateTenantsRequest.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package io.weaviate.client6.v1.api.collections.tenants;
22

3+
import java.util.ArrayList;
34
import java.util.Collections;
45
import java.util.List;
56

@@ -9,6 +10,35 @@
910
import io.weaviate.client6.v1.internal.rest.SimpleEndpoint;
1011

1112
public record UpdateTenantsRequest(List<Tenant> tenants) {
13+
/**
14+
* How many tenants the server accepts in a single update.
15+
*
16+
* <p>
17+
* {@code PUT /schema/{class}/tenants} rejects more than this with HTTP 422.
18+
* Note the asymmetry: adding tenants is not capped, only updating them, so
19+
* {@link CreateTenantsRequest} sends whatever it is given.
20+
*/
21+
static final int MAX_TENANTS_PER_REQUEST = 100;
22+
23+
/**
24+
* Split the tenants into requests the server will accept.
25+
*
26+
* <p>
27+
* The batches are views onto {@code tenants}, so the caller must not mutate it
28+
* while they are in flight.
29+
*/
30+
static List<List<Tenant>> batches(List<Tenant> tenants) {
31+
if (tenants.size() <= MAX_TENANTS_PER_REQUEST) {
32+
return List.of(tenants);
33+
}
34+
35+
var batches = new ArrayList<List<Tenant>>();
36+
for (int from = 0; from < tenants.size(); from += MAX_TENANTS_PER_REQUEST) {
37+
batches.add(tenants.subList(from, Math.min(from + MAX_TENANTS_PER_REQUEST, tenants.size())));
38+
}
39+
return batches;
40+
}
41+
1242
static Endpoint<UpdateTenantsRequest, Void> endpoint(CollectionDescriptor<?> collection) {
1343
return SimpleEndpoint.sideEffect(
1444
__ -> "PUT",

‎src/main/java/io/weaviate/client6/v1/api/collections/tenants/WeaviateTenantsClient.java‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,13 @@ public void update(Tenant... tenants) throws IOException {
119119
/**
120120
* Update tenant configuration.
121121
*
122+
* <p>
123+
* The server accepts at most
124+
* {@value UpdateTenantsRequest#MAX_TENANTS_PER_REQUEST} tenants per update, so
125+
* longer lists are sent as several sequential requests. That makes a partial
126+
* update possible: if one request fails, the tenants of the preceding ones stay
127+
* updated and the error is propagated.
128+
*
122129
* @param tenants Tenant names.
123130
* @throws WeaviateApiException in case the server returned with an
124131
* error status code.
@@ -127,7 +134,9 @@ public void update(Tenant... tenants) throws IOException {
127134
* or the server being unavailable.
128135
*/
129136
public void update(List<Tenant> tenants) throws IOException {
130-
this.restTransport.performRequest(new UpdateTenantsRequest(tenants), UpdateTenantsRequest.endpoint(collection));
137+
for (var batch : UpdateTenantsRequest.batches(tenants)) {
138+
this.restTransport.performRequest(new UpdateTenantsRequest(batch), UpdateTenantsRequest.endpoint(collection));
139+
}
131140
}
132141

133142
/**

‎src/main/java/io/weaviate/client6/v1/api/collections/tenants/WeaviateTenantsClientAsync.java‎

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -54,9 +54,24 @@ public CompletableFuture<Void> update(Tenant... tenants) throws IOException {
5454
return update(Arrays.asList(tenants));
5555
}
5656

57+
/**
58+
* Update tenant configuration.
59+
*
60+
* <p>
61+
* The server accepts at most
62+
* {@value UpdateTenantsRequest#MAX_TENANTS_PER_REQUEST} tenants per update, so
63+
* longer lists are sent as several requests, chained so that they reach the
64+
* server one after another. That makes a partial update possible: if one
65+
* request fails, the tenants of the preceding ones stay updated and the
66+
* returned future completes exceptionally.
67+
*/
5768
public CompletableFuture<Void> update(List<Tenant> tenants) throws IOException {
58-
return this.restTransport.performRequestAsync(new UpdateTenantsRequest(tenants),
59-
UpdateTenantsRequest.endpoint(collection));
69+
CompletableFuture<Void> updated = CompletableFuture.completedFuture(null);
70+
for (var batch : UpdateTenantsRequest.batches(tenants)) {
71+
updated = updated.thenCompose(__ -> this.restTransport.performRequestAsync(
72+
new UpdateTenantsRequest(batch), UpdateTenantsRequest.endpoint(collection)));
73+
}
74+
return updated;
6075
}
6176

6277
public CompletableFuture<Void> delete(String... tenants) throws IOException {
Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,121 @@
1+
package io.weaviate.client6.v1.api.collections.tenants;
2+
3+
import java.io.IOException;
4+
import java.util.ArrayList;
5+
import java.util.List;
6+
import java.util.stream.IntStream;
7+
8+
import org.assertj.core.api.Assertions;
9+
import org.junit.Test;
10+
11+
import io.weaviate.client6.v1.internal.orm.CollectionDescriptor;
12+
import io.weaviate.testutil.transport.MockRestTransport;
13+
14+
/**
15+
* The server rejects an update of more than 100 tenants with HTTP 422, so the
16+
* client splits longer lists. Adding tenants is not capped and has to keep going
17+
* out as a single request.
18+
*/
19+
public class UpdateTenantsBatchingTest {
20+
private static final CollectionDescriptor<?> COLLECTION = CollectionDescriptor.ofMap("Things");
21+
22+
private static List<Tenant> tenants(int n) {
23+
return IntStream.rangeClosed(1, n).mapToObj(i -> Tenant.active("tenant-" + i)).toList();
24+
}
25+
26+
private static List<String> names(int n) {
27+
return IntStream.rangeClosed(1, n).mapToObj(i -> "tenant-" + i).toList();
28+
}
29+
30+
/** Collect the body of every request the client sent, in order. */
31+
private static List<String> bodies(MockRestTransport transport, int expectAtMost) {
32+
var seen = new ArrayList<String>();
33+
var assertions = new MockRestTransport.AssertFunction[expectAtMost];
34+
for (var i = 0; i < expectAtMost; i++) {
35+
assertions[i] = (method, url, body, query) -> seen.add(body);
36+
}
37+
transport.assertNext(assertions);
38+
return seen;
39+
}
40+
41+
@Test
42+
public void test_batchesOfAtMost100() {
43+
Assertions.assertThat(UpdateTenantsRequest.batches(tenants(100)))
44+
.as("exactly at the limit is one request").hasSize(1);
45+
Assertions.assertThat(UpdateTenantsRequest.batches(tenants(101)))
46+
.as("one over the limit splits")
47+
.extracting(List::size).containsExactly(100, 1);
48+
Assertions.assertThat(UpdateTenantsRequest.batches(tenants(250)))
49+
.extracting(List::size).containsExactly(100, 100, 50);
50+
}
51+
52+
@Test
53+
public void test_batchesCoverEveryTenantInOrder() {
54+
var all = tenants(250);
55+
56+
var flattened = new ArrayList<Tenant>();
57+
UpdateTenantsRequest.batches(all).forEach(flattened::addAll);
58+
59+
Assertions.assertThat(flattened).isEqualTo(all);
60+
}
61+
62+
@Test
63+
public void test_emptyListIsASingleEmptyBatch() {
64+
Assertions.assertThat(UpdateTenantsRequest.batches(List.of())).containsExactly(List.of());
65+
}
66+
67+
@Test
68+
public void test_updateSplitsIntoSeveralRequests() throws IOException {
69+
var transport = new MockRestTransport();
70+
71+
new WeaviateTenantsClient(COLLECTION, transport, null).update(tenants(101));
72+
73+
var bodies = bodies(transport, 3);
74+
Assertions.assertThat(bodies).as("101 tenants -> 2 requests").hasSize(2);
75+
Assertions.assertThat(bodies.get(0)).contains("tenant-100").doesNotContain("tenant-101");
76+
Assertions.assertThat(bodies.get(1)).contains("tenant-101").doesNotContain("tenant-100");
77+
}
78+
79+
@Test
80+
public void test_updateAtTheLimitIsASingleRequest() throws IOException {
81+
var transport = new MockRestTransport();
82+
83+
new WeaviateTenantsClient(COLLECTION, transport, null).update(tenants(100));
84+
85+
Assertions.assertThat(bodies(transport, 2)).hasSize(1);
86+
}
87+
88+
/** activate/deactivate/offload all delegate to update, so they split too. */
89+
@Test
90+
public void test_deactivateSplitsIntoSeveralRequests() throws IOException {
91+
var transport = new MockRestTransport();
92+
93+
new WeaviateTenantsClient(COLLECTION, transport, null).deactivate(names(201));
94+
95+
var bodies = bodies(transport, 4);
96+
Assertions.assertThat(bodies).as("201 tenants -> 3 requests").hasSize(3);
97+
Assertions.assertThat(bodies).allSatisfy(body -> Assertions.assertThat(body).contains("INACTIVE"));
98+
}
99+
100+
@Test
101+
public void test_createIsNotSplit() throws IOException {
102+
var transport = new MockRestTransport();
103+
104+
new WeaviateTenantsClient(COLLECTION, transport, null).create(tenants(250));
105+
106+
var bodies = bodies(transport, 2);
107+
Assertions.assertThat(bodies).as("adding tenants is not capped by the server").hasSize(1);
108+
Assertions.assertThat(bodies.get(0)).contains("tenant-1").contains("tenant-250");
109+
}
110+
111+
@Test
112+
public void test_asyncUpdateSplitsIntoSeveralRequests() throws Exception {
113+
var transport = new MockRestTransport();
114+
115+
new WeaviateTenantsClientAsync(COLLECTION, transport, null).update(tenants(101)).get();
116+
117+
var bodies = bodies(transport, 3);
118+
Assertions.assertThat(bodies).hasSize(2);
119+
Assertions.assertThat(bodies.get(1)).contains("tenant-101");
120+
}
121+
}

‎src/test/java/io/weaviate/testutil/transport/MockRestTransport.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,9 @@ public <RequestT, ResponseT, ExceptionT> ResponseT performRequest(RequestT reque
5555
public <RequestT, ResponseT, ExceptionT> CompletableFuture<ResponseT> performRequestAsync(RequestT request,
5656
Endpoint<RequestT, ResponseT> endpoint) {
5757
requests.add(new Request<>(request, endpoint));
58-
return null;
58+
// A completed future rather than null, so callers which chain requests
59+
// (thenCompose) can be tested against this transport.
60+
return CompletableFuture.completedFuture(null);
5961
}
6062

6163
@Override

0 commit comments

Comments
 (0)