File tree Expand file tree Collapse file tree 1 file changed +2
-5
lines changed Expand file tree Collapse file tree 1 file changed +2
-5
lines changed Original file line number Diff line number Diff line change @@ -44,16 +44,13 @@ public <T> List<WatchResponse> chunkedPush(
4444) {
4545 List< WatchResponse> responses = new ArrayList<> ();
4646 List< T> records = new ArrayList<> ();
47- Int count = 0;
4847 Int waitBatchSize = batchSize / 10;
4948 if (waitBatchSize < 1) {
5049 waitBatchSize = batchSize;
5150 }
5251
5352 for (T item : objects) {
54- if (records.size() == batchSize || count == objects.size() - 1) {
55- count++;
56-
53+ if (records.size() == batchSize || ! item.hasNext()) {
5754 WatchResponse watch =
5855 this.push(
5956 indexName,
@@ -68,7 +65,7 @@ public <T> List<WatchResponse> chunkedPush(
6865
6966 records.add(item);
7067
71- if (waitForTasks && (responses.size() % waitBatchSize == 0 || count == objects.size() - 1 )) {
68+ if (waitForTasks && (responses.size() % waitBatchSize == 0 || !item.hasNext() )) {
7269 responses.subList(Math.max(responses.size() - waitBatchSize, 0), responses.size()).forEach(response -> {
7370 TaskUtils.retryUntil(
7471 () -> {
You can’t perform that action at this time.
0 commit comments