Replace BulkProcessor with our own implementation #2
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
@ewencp Ready for review. This PR replaces the BulkProcessor with our own implementation. The implementation did the following:
|
Ishiihara
force-pushed
the
bulk-processor
branch
from
ca0d458
to
0e08a15
Compare
| private static final long LINGER_MS_DEFAULT = 1; | ||
| private static final String LINGER_MS_DISPLAY = "Linger (ms)"; | ||
|
|
||
| public static final String MAX_INFLIGHT_REQUESTS_CONFIG = "max.in.flight.requests"; |
Ishiihara
force-pushed
the
bulk-processor
branch
6 times, most recently
from
9c21f43
to
283f8e9
Compare
|
@ewencp Ready for another round of review. Here are the major changes:
|
|
test this please |
| * @param maxInflightRequests The max allowed number of inflight requests. | ||
| * @return an instance of ElasticsearchWriter Builder. | ||
| */ | ||
| public Builder setMaxInflightRequests(int maxInflightRequests) { |
There was a problem hiding this comment.
nit, and not really that important since this is all internal apis, but since "in flight" is two words, I'd normally expect maxInFlightRequests instead of maxInflightRequests. (Same capitalization appears elsewhere too.)
Ishiihara
force-pushed
the
bulk-processor
branch
2 times, most recently
from
606ba13
to
27bc4d4
Compare
|
@ewencp Thanks for the review. Addressed the comments. PTAL. |
| } | ||
|
|
||
| private boolean canSubmit(RecordBatch batch, long now) { | ||
| if (guaranteeOrdering && muted) { |
There was a problem hiding this comment.
Let's leave this to a follow up patch since we've already got lots of updates in this one, but here's another idea. Could we just move retries into the execute() method? In the case of max in flight == 1, we just want to block on the first request until all retries are exhausted. With 1 executor, that should happen (as long as we can kill everything before any other submissions are processed). With max in flight > 1, we want to process multiple in parallel, but we know we won't have ordering guarantees, so we can just retry each as many times as needed (and we want each retry to be handled before anything behind it in the queue), and then give up and log an error. I think this approach would significantly simplify things?
|
@Ishiihara Left one final follow up re: delivery guarantees, but lets address it in follow up. LGTM. I do also see this transient failure, but rarely -- after running the tests 6 or 7 times. I think some of the messages are expected, but not sure which (if any) exceptions are related to the real issue: |
Ishiihara
force-pushed
the
bulk-processor
branch
from
27bc4d4
to
83be437
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
No description provided.