Skip to content

KAFKA-16770; [2/2] Coalesce records into bigger batches #16215

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dajac merged 13 commits into from
Jun 12, 2024
Merged

KAFKA-16770; [2/2] Coalesce records into bigger batches #16215

Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
f140494
add config
dajac Jun 4, 2024
590c8e4
reorganize code
dajac Jun 4, 2024
5139390
add batching
dajac Jun 4, 2024
e02f118
address comments
dajac Jun 6, 2024
187defa
add logs
dajac Jun 6, 2024
ab3706b
refactor
dajac Jun 6, 2024
aaf4d35
tests + refactors
dajac Jun 6, 2024
1d689f1
wire config correctly
dajac Jun 6, 2024
45d6175
address comments
dajac Jun 10, 2024
c7bbf99
fail and reload the state machine
dajac Jun 10, 2024
929aba8
Merge remote-tracking branch 'upstream/trunk' into KAFKA-16770-2
dajac Jun 10, 2024
cfafdcb
add test + small fix
dajac Jun 11, 2024
e958fd3
address comment
dajac Jun 11, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 51 additions & 32 deletions ...rdinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -638,6 +638,7 @@ private void transitionTo(
.build(),
tp
);
load();
break;

case ACTIVE:
Expand All @@ -664,6 +665,46 @@ private void transitionTo(
runtimeMetrics.recordPartitionStateChange(oldState, state);
}

/**
* Loads the coordinator.
*/
private void load() {
if (state != CoordinatorState.LOADING) {
throw new IllegalStateException("Coordinator must be in loading state");
}

loader.load(tp, coordinator).whenComplete((summary, exception) -> {
scheduleInternalOperation("CompleteLoad(tp=" + tp + ", epoch=" + epoch + ")", tp, () -> {
CoordinatorContext context = coordinators.get(tp);
if (context != null) {
if (context.state != CoordinatorState.LOADING) {
log.info("Ignored load completion from {} because context is in {} state.",
context.tp, context.state);
return;
}
try {
if (exception != null) throw exception;
context.transitionTo(CoordinatorState.ACTIVE);
if (summary != null) {
runtimeMetrics.recordPartitionLoadSensor(summary.startTimeMs(), summary.endTimeMs());
log.info("Finished loading of metadata from {} with epoch {} in {}ms where {}ms " +
"was spent in the scheduler. Loaded {} records which total to {} bytes.",
tp, epoch, summary.endTimeMs() - summary.startTimeMs(),
summary.schedulerQueueTimeMs(), summary.numRecords(), summary.numBytes());
}
} catch (Throwable ex) {
log.error("Failed to load metadata from {} with epoch {} due to {}.",
tp, epoch, ex.toString());
context.transitionTo(CoordinatorState.FAILED);
}
} else {
log.debug("Failed to complete the loading of metadata for {} in epoch {} since the coordinator does not exist.",
tp, epoch);
}
});
});
}

/**
* Unloads the coordinator.
*/
Expand Down Expand Up @@ -711,8 +752,16 @@ private void flushCurrentBatch() {
coordinator.updateLastWrittenOffset(offset);

if (offset != currentBatch.nextOffset) {
throw new IllegalStateException("The state machine of coordinator " + tp + " is out of sync with the " +
"underlying log. The last write returned " + offset + " while " + currentBatch.nextOffset + " was expected");
log.error("The state machine of the coordinator {} is out of sync with the underlying log. " +
"The last written offset returned is {} while the coordinator expected {}. The coordinator " +
"will be reloaded in order to re-synchronize the state machine.",
tp, offset, currentBatch.nextOffset);
// Transition to FAILED state to unload the state machine and complete
// exceptionally all the pending operations.
transitionTo(CoordinatorState.FAILED);
// Transition to LOADING to trigger the restoration of the state.
transitionTo(CoordinatorState.LOADING);
Comment thread
dajac marked this conversation as resolved.
return;
}

// Add all the pending deferred events to the deferred event queue.
Expand Down Expand Up @@ -2164,36 +2213,6 @@ public void scheduleLoadOperation(
case FAILED:
case INITIAL:
context.transitionTo(CoordinatorState.LOADING);
loader.load(tp, context.coordinator).whenComplete((summary, exception) -> {
scheduleInternalOperation("CompleteLoad(tp=" + tp + ", epoch=" + partitionEpoch + ")", tp, () -> {
CoordinatorContext ctx = coordinators.get(tp);
if (ctx != null) {
if (ctx.state != CoordinatorState.LOADING) {
log.info("Ignored load completion from {} because context is in {} state.",
ctx.tp, ctx.state);
return;
}
try {
if (exception != null) throw exception;
ctx.transitionTo(CoordinatorState.ACTIVE);
if (summary != null) {
runtimeMetrics.recordPartitionLoadSensor(summary.startTimeMs(), summary.endTimeMs());
log.info("Finished loading of metadata from {} with epoch {} in {}ms where {}ms " +
"was spent in the scheduler. Loaded {} records which total to {} bytes.",
tp, partitionEpoch, summary.endTimeMs() - summary.startTimeMs(),
summary.schedulerQueueTimeMs(), summary.numRecords(), summary.numBytes());
}
} catch (Throwable ex) {
log.error("Failed to load metadata from {} with epoch {} due to {}.",
tp, partitionEpoch, ex.toString());
ctx.transitionTo(CoordinatorState.FAILED);
}
} else {
log.debug("Failed to complete the loading of metadata for {} in epoch {} since the coordinator does not exist.",
tp, partitionEpoch);
}
});
});
break;

case LOADING:
Expand Down