Skip to content

Commit 4b81a79

Browse files
l-trottal-trotta
committed
more refactoring
1 parent 69842ee commit 4b81a79

File tree

3 files changed

+19
-31
lines changed

3 files changed

+19
-31
lines changed

‎java-client/src/main/java/co/elastic/clients/elasticsearch/_helpers/bulk/BulkIngester.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -536,7 +536,7 @@ private void innerAdd(BulkOperationRepeatable<Context> repeatableOp) {
536536
IngesterOperation ingestOp = IngesterOperation.of(repeatableOp, client._jsonpMapper());
537537

538538
addCondition.whenReady(() -> {
539-
operations.add(ingestOp.operation());
539+
operations.add(ingestOp.repeatableOperation());
540540
currentSize += ingestOp.size();
541541

542542
if (!canAddOperation()) {

‎java-client/src/main/java/co/elastic/clients/elasticsearch/_helpers/bulk/IngesterOperation.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,7 @@ public static IngesterOperation of(BulkOperationRepeatable repeatableOp, JsonpMa
5858
}
5959
}
6060

61-
public BulkOperationRepeatable operation() {
61+
public BulkOperationRepeatable repeatableOperation() {
6262
return this.repeatableOp;
6363
}
6464

‎java-client/src/test/java/co/elastic/clients/elasticsearch/_helpers/bulk/BulkIngesterTest.java

Lines changed: 17 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@
3232
import co.elastic.clients.json.JsonpMapper;
3333
import co.elastic.clients.json.JsonpUtils;
3434
import co.elastic.clients.json.SimpleJsonpMapper;
35-
import co.elastic.clients.transport.BackoffPolicy;
3635
import co.elastic.clients.transport.ElasticsearchTransport;
3736
import co.elastic.clients.transport.Endpoint;
3837
import co.elastic.clients.transport.TransportOptions;
@@ -142,7 +141,7 @@ private void multiThreadTest(int maxOperations, int maxRequests, int numThreads,
142141
for (int i = 0; i < numThreads; i++) {
143142
new Thread(() -> {
144143
try {
145-
Thread.sleep((long) (Math.random() * 100));
144+
Thread.sleep((long)(Math.random() * 100));
146145
} catch (InterruptedException e) {
147146
throw new RuntimeException(e);
148147
}
@@ -169,8 +168,7 @@ private void multiThreadTest(int maxOperations, int maxRequests, int numThreads,
169168
assertEquals(expectedOperations, listener.operations.get());
170169
assertEquals(expectedOperations, transport.operations.get());
171170

172-
int expectedRequests =
173-
expectedOperations / maxOperations + ((expectedOperations % maxOperations == 0) ? 0 : 1);
171+
int expectedRequests = expectedOperations / maxOperations + ((expectedOperations % maxOperations == 0) ? 0 : 1) ;
174172

175173
assertEquals(expectedRequests, ingester.requestCount());
176174
assertEquals(expectedRequests, listener.requests.get());
@@ -185,8 +183,7 @@ public void multiThreadStressTest() throws InterruptedException, IOException {
185183

186184
// DISCLAIMER: this configuration is highly inefficient and only used here to showcase an extreme
187185
// situation where the number of adding threads greatly exceeds the number of concurrent requests
188-
// handled by the ingester. It's strongly recommended to always tweak maxConcurrentRequests
189-
// accordingly.
186+
// handled by the ingester. It's strongly recommended to always tweak maxConcurrentRequests accordingly.
190187
BulkIngester<?> ingester = BulkIngester.of(b -> b
191188
.client(client)
192189
.globalSettings(s -> s.index(index))
@@ -212,22 +209,21 @@ public void multiThreadStressTest() throws InterruptedException, IOException {
212209
executor.submit(thread);
213210
}
214211

215-
executor.awaitTermination(10, TimeUnit.SECONDS);
212+
executor.awaitTermination(10,TimeUnit.SECONDS);
216213
ingester.close();
217214

218215
client.indices().refresh();
219216

220217
IndicesStatsResponse indexStats = client.indices().stats(g -> g.index(index));
221218

222-
assertTrue(indexStats.indices().get(index).primaries().docs().count() == 100000);
219+
assertTrue(indexStats.indices().get(index).primaries().docs().count()==100000);
223220
}
224221

225222
@Test
226223
public void sizeLimitTest() throws Exception {
227224
TestTransport transport = new TestTransport();
228225

229-
long operationSize = IngesterOperation.of(new BulkOperationRepeatable<>(operation, null, null),
230-
transport.jsonpMapper()).size();
226+
long operationSize = IngesterOperation.of(new BulkOperationRepeatable<>(operation, null, null), transport.jsonpMapper()).size();
231227

232228
BulkIngester<?> ingester = BulkIngester.of(b -> b
233229
.client(new ElasticsearchAsyncClient(transport))
@@ -257,7 +253,7 @@ public void periodicFlushTest() throws Exception {
257253
// Disable other flushing limits
258254
.maxSize(-1)
259255
.maxOperations(-1)
260-
.maxConcurrentRequests(Integer.MAX_VALUE - 1)
256+
.maxConcurrentRequests(Integer.MAX_VALUE-1)
261257
);
262258

263259
// Add an operation every 100 ms to give time
@@ -297,8 +293,7 @@ public void beforeBulk(long executionId, BulkRequest request, List<Void> context
297293
}
298294

299295
@Override
300-
public void afterBulk(long executionId, BulkRequest request, List<Void> contexts,
301-
BulkResponse response) {
296+
public void afterBulk(long executionId, BulkRequest request, List<Void> contexts, BulkResponse response) {
302297
if (executionId == 2) {
303298
// Fail after the request is sent
304299
failureCount.incrementAndGet();
@@ -307,8 +302,7 @@ public void afterBulk(long executionId, BulkRequest request, List<Void> contexts
307302
}
308303

309304
@Override
310-
public void afterBulk(long executionId, BulkRequest request, List<Void> contexts,
311-
Throwable failure) {
305+
public void afterBulk(long executionId, BulkRequest request, List<Void> contexts, Throwable failure) {
312306

313307
}
314308
};
@@ -361,13 +355,11 @@ public void beforeBulk(long executionId, BulkRequest request, List<Integer> cont
361355
}
362356

363357
@Override
364-
public void afterBulk(long executionId, BulkRequest request, List<Integer> contexts,
365-
BulkResponse response) {
358+
public void afterBulk(long executionId, BulkRequest request, List<Integer> contexts, BulkResponse response) {
366359
}
367360

368361
@Override
369-
public void afterBulk(long executionId, BulkRequest request, List<Integer> contexts,
370-
Throwable failure) {
362+
public void afterBulk(long executionId, BulkRequest request, List<Integer> contexts, Throwable failure) {
371363
}
372364
};
373365

@@ -381,7 +373,7 @@ public void afterBulk(long executionId, BulkRequest request, List<Integer> conte
381373
for (int i = 0; i < 10; i++) {
382374
for (int j = 0; j < 10; j++) {
383375
// Set a context only after 5, so that we test filling with nulls.
384-
Integer context = j < 5 ? null : i * 10 + j;
376+
Integer context = j < 5 ? null : i*10 + j;
385377
ingester.add(operation, context);
386378
}
387379
}
@@ -399,7 +391,7 @@ public void afterBulk(long executionId, BulkRequest request, List<Integer> conte
399391
if (j < 5) {
400392
assertNull(contexts.get(j));
401393
} else {
402-
assertEquals(contexts.get(j), i * 10 + j);
394+
assertEquals(contexts.get(j), i*10 + j);
403395
}
404396
}
405397
}
@@ -441,8 +433,7 @@ public void beforeBulk(long executionId, BulkRequest request, List<Void> context
441433

442434
@Test
443435
public void pipelineTest() {
444-
String json = "{\"create\":{\"_id\":\"some_id\",\"_index\":\"some_idx\",\"pipeline\":\"pipe\"," +
445-
"\"require_alias\":true}}";
436+
String json = "{\"create\":{\"_id\":\"some_id\",\"_index\":\"some_idx\",\"pipeline\":\"pipe\",\"require_alias\":true}}";
446437
JsonpMapper mapper = new SimpleJsonpMapper();
447438

448439
BulkOperation create = BulkOperation.of(o -> o.create(c -> c
@@ -456,8 +447,7 @@ public void pipelineTest() {
456447
String createStr = JsonpUtils.toJsonString(create, mapper);
457448
assertEquals(json, createStr);
458449

459-
BulkOperation create1 = IngesterOperation.of(new BulkOperationRepeatable<>(create, null, null),
460-
mapper).operation().getOperation();
450+
BulkOperation create1 = IngesterOperation.of(new BulkOperationRepeatable<>(create, null, null), mapper).repeatableOperation().getOperation();
461451

462452
String create1Str = JsonpUtils.toJsonString(create1, mapper);
463453
assertEquals(json, create1Str);
@@ -546,15 +536,13 @@ public void testConfigValidation() {
546536
private static class CountingListener implements BulkListener<Void> {
547537
public final AtomicInteger operations = new AtomicInteger();
548538
public final AtomicInteger requests = new AtomicInteger();
549-
550539
@Override
551540
public void beforeBulk(long executionId, BulkRequest request, List<Void> contexts) {
552541

553542
}
554543

555544
@Override
556-
public void afterBulk(long executionId, BulkRequest request, List<Void> contexts,
557-
BulkResponse response) {
545+
public void afterBulk(long executionId, BulkRequest request, List<Void> contexts, BulkResponse response) {
558546
operations.addAndGet(request.operations().size());
559547
requests.incrementAndGet();
560548
}
@@ -607,7 +595,7 @@ public <RequestT, ResponseT, ErrorT> CompletableFuture<ResponseT> performRequest
607595
});
608596

609597
@SuppressWarnings("unchecked")
610-
CompletableFuture<ResponseT> result = (CompletableFuture<ResponseT>) response;
598+
CompletableFuture<ResponseT> result = (CompletableFuture<ResponseT>)response;
611599
return result;
612600
}
613601

0 commit comments

Comments
 (0)