MINOR: Remove throttling logic from RecordAccumulator

[ijuma]

This is redundant since Sender and NetworkClient handle throttling. It's
also confusing since the RecordAccumulator logic only applies when
max.in.flight.requests.per.connection=1.

In Sender.sendProducerData, the following code handles throttling:

while (iter.hasNext()) {
    Node node = iter.next();
    if (!this.client.ready(node, now)) {
        iter.remove();
        notReadyTimeout = Math.min(notReadyTimeout, this.client.pollDelayMs(node, now));
    }
}

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[ijuma]

One thing I don't understand is why this logic only runs if max.in.flight=1. cc @lbradstreet

[lbradstreet]

Is the idea that we can remove the throttling here, and rely on the existing NetworkClient throttle instead?

[ijuma]

Yes. This logic was only active for max.inflight=1 which is confusing and seemingly unnecessary. We have logic to remove nodes from readyNodes if the client is not ready.

[lbradstreet]

Sounds good. It does look pretty unnecessary to me too.

[ijuma]

Will fix the checkstyle error.

[ijuma]

@junrao Does this look good to you?

[ijuma]

Retest this please. One build passed, two timed out.

[ijuma]

One job passed and the other failed with two known flakes:

kafka.admin.ResetConsumerGroupOffsetTest.testResetOffsetsExportImportPlan
org.apache.kafka.connect.mirror.MirrorConnectorsIntegrationTest.testReplication

@rajinisivaram does this look good to you?

[rajinisivaram]

@ijuma Thanks for the PR, LGTM