Skip to content

KAFKA-8768: DeleteRecords request/response automated protocol#7957

Merged
mimaison merged 3 commits intoapache:trunkfrom
tombentley:KAFKA-8768-DeleteRecords-automated-protocol
Mar 13, 2020
Merged

KAFKA-8768: DeleteRecords request/response automated protocol#7957
mimaison merged 3 commits intoapache:trunkfrom
tombentley:KAFKA-8768-DeleteRecords-automated-protocol

Conversation

@tombentley
Copy link
Copy Markdown
Member

@tombentley tombentley commented Jan 14, 2020

Also add version 2 to make use of flexible versions, per KIP-482.

Committer Checklist (excluded from commit message)

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

@tombentley
Copy link
Copy Markdown
Member Author

tombentley commented Jan 14, 2020

@mimaison

@tombentley tombentley force-pushed the KAFKA-8768-DeleteRecords-automated-protocol branch from 94ae971 to 9512f79 Compare January 14, 2020 14:01
@tombentley
Copy link
Copy Markdown
Member Author

tombentley commented Jan 17, 2020

@cmccabe are you able to review this?

@cmccabe
Copy link
Copy Markdown
Contributor

cmccabe commented Jan 18, 2020

I'm going to be on vacation the next two weeks, but I'll certainly review after that. Sorry for the delays

@mimaison
Copy link
Copy Markdown
Member

mimaison commented Jan 30, 2020

@tombentley Can you rebase on trunk?

@tombentley tombentley force-pushed the KAFKA-8768-DeleteRecords-automated-protocol branch from 9512f79 to 18c4330 Compare February 4, 2020 09:25
@tombentley
Copy link
Copy Markdown
Member Author

tombentley commented Feb 4, 2020

@mimaison done

mimaison
mimaison requested changes Feb 6, 2020
View reviewed changes
Copy link
Copy Markdown
Member

@mimaison mimaison left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @tombentley
It looks good overall, I've left a few comments

Comment thread clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java Outdated
Comment thread clients/src/main/java/org/apache/kafka/common/requests/DeleteRecordsRequest.java Outdated
Comment thread core/src/main/scala/kafka/server/KafkaApis.scala Outdated
Comment thread core/src/main/scala/kafka/server/KafkaApis.scala Outdated
Comment thread core/src/main/scala/kafka/server/KafkaApis.scala Outdated
@mimaison mimaison mentioned this pull request Feb 6, 2020
Merged
3 tasks
@tombentley
KAFKA-8768: Replace DeleteRecords request/response with automated pro…
71b6b46 
…tocol

Also add version 2 to make use of flexible versions, per KIP-482.
@tombentley
Fix comments
8861774
@tombentley tombentley force-pushed the KAFKA-8768-DeleteRecords-automated-protocol branch from 18c4330 to 8861774 Compare February 25, 2020 11:35
@tombentley
Copy link
Copy Markdown
Member Author

tombentley commented Feb 25, 2020

@mimaison fixed, if you want to take another look.

@tombentley
Copy link
Copy Markdown
Member Author

tombentley commented Feb 27, 2020

ok to test

mimaison
mimaison requested changes Mar 13, 2020
View reviewed changes
Copy link
Copy Markdown
Member

@mimaison mimaison left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the update. It looks good overall.
I've left a few suggestions/questions for minor issues

Comment thread clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java Outdated
if (node != null) {
leaders.computeIfAbsent(node, key -> new HashMap<>()).put(entry.getKey(),
entry.getValue().beforeOffset());
if (!leaders.containsKey(node))
Copy link
Copy Markdown
Member

@mimaison mimaison Mar 13, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we can computeIfAbsent() here. It will replace containsKey, put and get.

Comment thread clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
if (!leaders.containsKey(node))
leaders.put(node, new HashMap<>());
Map<String, DeleteRecordsTopic> deletionsForLeader = leaders.get(node);
DeleteRecordsTopic deleteRecords = deletionsForLeader.get(topicPartition.topic());
Copy link
Copy Markdown
Member

@mimaison mimaison Mar 13, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same here, I think computeIfAbsent would simplify the logic

Comment thread clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java Outdated
partitionDeleteOffsets.keySet().stream().map(futures::get);
partitionDeleteOffsets.values().stream().flatMap(
recordsToDelete1 -> {
Stream<TopicPartition> topicPartitionStream = recordsToDelete1.partitions().stream().map(partitionsToDelete ->
Copy link
Copy Markdown
Member

@mimaison mimaison Mar 13, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we return that directly without creating a variable?

Comment thread clients/src/main/java/org/apache/kafka/common/requests/DeleteRecordsResponse.java Outdated

private final int throttleTimeMs;
private final Map<TopicPartition, PartitionResponse> responses;
public static final long INVALID_LOW_WATERMARK = -1;
Copy link
Copy Markdown
Member

@mimaison mimaison Mar 13, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's keep the L

Comment thread core/src/main/scala/kafka/server/KafkaApis.scala Outdated
deleteRecordsRequest.partitionOffsets.asScala.toSeq.map(_._1.topic))
for ((topicPartition, offset) <- deleteRecordsRequest.partitionOffsets.asScala) {
deleteRecordsRequest.data.topics.asScala.map(_.name))
for ((topicPartition, offset) <- deleteRecordsRequest.data.topics.asScala.flatMap(deleteTopic => {
Copy link
Copy Markdown
Member

@mimaison mimaison Mar 13, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder if it's more readable having 2 nested for loops. What do you think?

Copy link
Copy Markdown
Member Author

@tombentley tombentley Mar 13, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To me what makes it less readable is that the expression for the for loop is the flatMap & map. I think the simplest was to improve that would just be to factor out a val for that expression. Does that work for you?

Comment thread core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala Outdated
ApiKeys.DELETE_TOPICS -> ((resp: requests.DeleteTopicsResponse) => Errors.forCode(resp.data.responses.find(topic).errorCode())),
ApiKeys.DELETE_RECORDS -> ((resp: requests.DeleteRecordsResponse) => resp.responses.get(tp).error),
ApiKeys.DELETE_RECORDS -> ((resp: requests.DeleteRecordsResponse) => Errors.forCode(resp.data.topics
.find(tp.topic).partitions().find(tp.partition).errorCode)),
Copy link
Copy Markdown
Member

@mimaison mimaison Mar 13, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No need for brackets after partitions

@tombentley
Copy link
Copy Markdown
Member Author

tombentley commented Mar 13, 2020

@mimaison I've pushed fixes and tested locally.

mimaison
mimaison approved these changes Mar 13, 2020
View reviewed changes
Copy link
Copy Markdown
Member

@mimaison mimaison left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM, thanks for the quick update

@mimaison
Copy link
Copy Markdown
Member

mimaison commented Mar 13, 2020

All tests passed, merging to trunk

@mimaison mimaison merged commit f869e33 into apache:trunk Mar 13, 2020
ijuma added a commit to confluentinc/kafka that referenced this pull request Mar 17, 2020
@ijuma
Merge remote-tracking branch 'apache-github/trunk' into ccs-master
6f7084a 
* apache-github/trunk: (39 commits)
  MINOR: cleanup and add tests to StateDirectoryTest (apache#8304)
  HOTFIX: StateDirectoryTest should use Set instead of List (apache#8305)
  MINOR: Fix build and JavaDoc warnings (apache#8291)
  MINOR: Fix kafka.server.RequestQuotaTest missing new ApiKeys. (apache#8302)
  KAFKA-9712: Catch and handle exception thrown by reflections scanner (apache#8289)
  KAFKA-9670; Reduce allocations in Metadata Response preparation (apache#8236)
  MINOR: fix Scala 2.13 build error introduced in apache#8083 (apache#8301)
  MINOR: enforce non-negative invariant for checkpointed offsets (apache#8297)
  MINOR: comment apikey types in generated switch (apache#8201)
  MINOR: Fix typo in CreateTopicsResponse.json (apache#8300)
  KIP-546: Implement describeClientQuotas and alterClientQuotas. (apache#8083)
  KAFKA-6647: Do note delete the lock file while holding the lock (apache#8267)
  KAFKA-9677: Fix consumer fetch with small consume bandwidth quotas (apache#8290)
  KAFKA-9533: Fix JavaDocs of KStream.transformValues (apache#8298)
  MINOR: reuse pseudo-topic in FKJoin (apache#8296)
  KAFKA-6145: Pt 2. Include offset sums in subscription (apache#8246)
  KAFKA-9714; Eliminate unused reference to IBP in `TransactionStateManager` (apache#8293)
  KAFKA-9718; Don't log passwords for AlterConfigs in request logs (apache#8294)
  KAFKA-8768: DeleteRecords request/response automated protocol (apache#7957)
  KAFKA-9685: Solve Set concatenation perf issue in AclAuthorizer
  ...
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@mimaison mimaison mimaison approved these changes

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@tombentley @cmccabe @mimaison