lianetm commented on code in PR #23749:
URL: https://github.com/apache/kafka/pull/23749#discussion_r4298188222
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -1424,25 +1406,63 @@ public static RecordAppendResult
appended(FutureRecordMetadata future,
* bytes of chunk capacity and retry. The append was not attempted.
*/
public static RecordAppendResult needsExtension(int
extensionBytesNeeded) {
- return new RecordAppendResult(Outcome.NEEDS_BUFFER_EXTENSION,
null, false, false, extensionBytesNeeded, 0);
+ return new BufferExtensionResult(extensionBytesNeeded);
}
/**
* @return {@code true} if the record was appended to the open batch
({@link #future} is then
* non-null), {@code false} for the {@link #needsBufferExtension()}
and {@link #needsNewBatch()} results.
*/
public boolean appended() {
- return outcome == Outcome.APPENDED;
+ return future != null;
}
/** @return {@code true} if the open batch needs more chunk capacity
before the record fits. */
public boolean needsBufferExtension() {
- return outcome == Outcome.NEEDS_BUFFER_EXTENSION;
+ return false;
}
/** @return {@code true} if there is no open batch that can take the
record, so a new one must be created. */
public boolean needsNewBatch() {
- return outcome == Outcome.NEEDS_NEW_BATCH;
+ return future == null;
+ }
+
+ /**
+ * Return the chunk capacity required by a buffer-extension result.
+ *
+ * @return the number of bytes to allocate
+ * @throws IllegalStateException if this is not a buffer-extension
result
+ */
+ int extensionBytesNeeded() {
+ throw new IllegalStateException("Only a buffer-extension result
has extension bytes");
+ }
+
+ /**
+ * The only result that carries incremental-strategy-specific state.
Keeping this state
+ * out of the regular result avoids allocating space for it on every
full-strategy append.
Review Comment:
Should we clarify first what this represents? (the open batch needs more
chunk capacity to append the record)
Then we can leave the info about it being incremental-only and why separate.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -1424,25 +1406,63 @@ public static RecordAppendResult
appended(FutureRecordMetadata future,
* bytes of chunk capacity and retry. The append was not attempted.
*/
public static RecordAppendResult needsExtension(int
extensionBytesNeeded) {
- return new RecordAppendResult(Outcome.NEEDS_BUFFER_EXTENSION,
null, false, false, extensionBytesNeeded, 0);
+ return new BufferExtensionResult(extensionBytesNeeded);
}
/**
* @return {@code true} if the record was appended to the open batch
({@link #future} is then
* non-null), {@code false} for the {@link #needsBufferExtension()}
and {@link #needsNewBatch()} results.
*/
public boolean appended() {
- return outcome == Outcome.APPENDED;
+ return future != null;
}
/** @return {@code true} if the open batch needs more chunk capacity
before the record fits. */
public boolean needsBufferExtension() {
- return outcome == Outcome.NEEDS_BUFFER_EXTENSION;
+ return false;
}
/** @return {@code true} if there is no open batch that can take the
record, so a new one must be created. */
public boolean needsNewBatch() {
- return outcome == Outcome.NEEDS_NEW_BATCH;
+ return future == null;
+ }
+
+ /**
+ * Return the chunk capacity required by a buffer-extension result.
Review Comment:
```suggestion
* Return the chunk capacity required by a needsBufferExtension
result.
```
same for other usages of "buffer-extension", I would replace it with the
known func needsBufferExtension just to be consistent
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
