[GitHub] [kafka] showuon commented on a change in pull request #11460: KAFKA-13425: Optimization of KafkaConsumer#pause semantics
showuon commented on a change in pull request #11460: URL: https://github.com/apache/kafka/pull/11460#discussion_r742575556 ## File path: clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java ## @@ -2007,7 +2007,7 @@ public OffsetAndMetadata committed(TopicPartition partition, final Duration time * Suspend fetching from the requested partitions. Future calls to {@link #poll(Duration)} will not return * any records from these partitions until they have been resumed using {@link #resume(Collection)}. * Note that this method does not affect partition subscription. In particular, it does not cause a group - * rebalance when automatic assignment is used. + * rebalance when automatic assignment is used. And groupRebalance does not preserve pause state. Review comment: How about this: ``` * ... * rebalance when automatic assignment is used. * * Note: Rebalance will not preserve the pause/resume state. ``` ## File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java ## @@ -307,6 +307,10 @@ private Exception invokePartitionsAssigned(final Set assignedPar private Exception invokePartitionsRevoked(final Set revokedPartitions) { log.info("Revoke previously assigned partitions {}", Utils.join(revokedPartitions, ", ")); +Set revokePausedPartitions = subscriptions.pausedPartitions(); +revokePausedPartitions.retainAll(revokedPartitions); +if (!revokePausedPartitions.isEmpty()) +log.info("Revoke previously paused partitions {}", Utils.join(revokePausedPartitions, ", ")); Review comment: I'm afraid that users might get confused, because it looks like we revoke twice: ``` Revoke previously assigned partitions tp1, tp2. Revoke previously paused partitions tp1 ``` How about it: `log.info("The pause flag in partitions [{}] will be removed due to revocation.", Utils.join(revokePausedPartitions, ", "));` WDYT? -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] RivenSun2 commented on a change in pull request #11460: KAFKA-13425: Optimization of KafkaConsumer#pause semantics
RivenSun2 commented on a change in pull request #11460: URL: https://github.com/apache/kafka/pull/11460#discussion_r742589553 ## File path: clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java ## @@ -2007,7 +2007,7 @@ public OffsetAndMetadata committed(TopicPartition partition, final Duration time * Suspend fetching from the requested partitions. Future calls to {@link #poll(Duration)} will not return * any records from these partitions until they have been resumed using {@link #resume(Collection)}. * Note that this method does not affect partition subscription. In particular, it does not cause a group - * rebalance when automatic assignment is used. + * rebalance when automatic assignment is used. And groupRebalance does not preserve pause state. Review comment: Sure -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] RivenSun2 commented on a change in pull request #11460: KAFKA-13425: Optimization of KafkaConsumer#pause semantics
RivenSun2 commented on a change in pull request #11460: URL: https://github.com/apache/kafka/pull/11460#discussion_r742589702 ## File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java ## @@ -307,6 +307,10 @@ private Exception invokePartitionsAssigned(final Set assignedPar private Exception invokePartitionsRevoked(final Set revokedPartitions) { log.info("Revoke previously assigned partitions {}", Utils.join(revokedPartitions, ", ")); +Set revokePausedPartitions = subscriptions.pausedPartitions(); +revokePausedPartitions.retainAll(revokedPartitions); +if (!revokePausedPartitions.isEmpty()) +log.info("Revoke previously paused partitions {}", Utils.join(revokePausedPartitions, ", ")); Review comment: I agree with you -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] RivenSun2 commented on pull request #11460: KAFKA-13425: Optimization of KafkaConsumer#pause semantics
RivenSun2 commented on pull request #11460: URL: https://github.com/apache/kafka/pull/11460#issuecomment-960521298 @showuon Thanks for your review I just commited a new change -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11460: KAFKA-13425: Optimization of KafkaConsumer#pause semantics
showuon commented on a change in pull request #11460: URL: https://github.com/apache/kafka/pull/11460#discussion_r742596144 ## File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java ## @@ -326,6 +330,10 @@ private Exception invokePartitionsRevoked(final Set revokedParti private Exception invokePartitionsLost(final Set lostPartitions) { log.info("Lost previously assigned partitions {}", Utils.join(lostPartitions, ", ")); +Set lostPausedPartitions = subscriptions.pausedPartitions(); +lostPausedPartitions.retainAll(lostPartitions); +if (!lostPausedPartitions.isEmpty()) +log.info("The pause flag in partitions [{}] will be removed due to lost.", Utils.join(lostPausedPartitions, ", ")); Review comment: nit: due to **partition** lost -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] RivenSun2 commented on pull request #11340: KAFKA-13310 : KafkaConsumer cannot jump out of the poll method, and the…
RivenSun2 commented on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-960522880 @guozhangwang @showuon Thanks for your helpful review. I just commit a new change If you have time, please browse it. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] RivenSun2 commented on a change in pull request #11460: KAFKA-13425: Optimization of KafkaConsumer#pause semantics
RivenSun2 commented on a change in pull request #11460: URL: https://github.com/apache/kafka/pull/11460#discussion_r742601310 ## File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java ## @@ -326,6 +330,10 @@ private Exception invokePartitionsRevoked(final Set revokedParti private Exception invokePartitionsLost(final Set lostPartitions) { log.info("Lost previously assigned partitions {}", Utils.join(lostPartitions, ", ")); +Set lostPausedPartitions = subscriptions.pausedPartitions(); +lostPausedPartitions.retainAll(lostPartitions); +if (!lostPausedPartitions.isEmpty()) +log.info("The pause flag in partitions [{}] will be removed due to lost.", Utils.join(lostPausedPartitions, ", ")); Review comment: Thank you for your rigorous attitude. Forgive my bad English. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Created] (KAFKA-13433) JsonConverter's method convertToJson when field is optional with default value and value is null, return default value.
Aiden Gong created KAFKA-13433: -- Summary: JsonConverter's method convertToJson when field is optional with default value and value is null, return default value. Key: KAFKA-13433 URL: https://issues.apache.org/jira/browse/KAFKA-13433 Project: Kafka Issue Type: Bug Components: KafkaConnect Affects Versions: 3.0.0 Reporter: Aiden Gong Attachments: image-2021-11-04-16-25-18-890.png, image-2021-11-04-16-25-18-975.png JsonConverter's method convertToJson when field is optional with default value and value is null, return default value. !image-2021-11-04-16-25-18-975.png! -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-13433) JsonConverter's method convertToJson when field is optional with default value and value is null, return default value.
[ https://issues.apache.org/jira/browse/KAFKA-13433?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Aiden Gong updated KAFKA-13433: --- Description: JsonConverter's method convertToJson when field is optional with default value and value is null, return default value. (was: JsonConverter's method convertToJson when field is optional with default value and value is null, return default value. !image-2021-11-04-16-25-18-975.png!) > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. > > > Key: KAFKA-13433 > URL: https://issues.apache.org/jira/browse/KAFKA-13433 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Affects Versions: 3.0.0 >Reporter: Aiden Gong >Priority: Blocker > Attachments: image-2021-11-04-16-25-18-890.png, > image-2021-11-04-16-25-18-975.png > > > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-13433) JsonConverter's method convertToJson when field is optional with default value and value is null, return default value.
[ https://issues.apache.org/jira/browse/KAFKA-13433?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Aiden Gong updated KAFKA-13433: --- Description: JsonConverter's method convertToJson when field is optional with default value and value is null, return default value. Please refer attachments for more detail. (was: JsonConverter's method convertToJson when field is optional with default value and value is null, return default value. ) > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. > > > Key: KAFKA-13433 > URL: https://issues.apache.org/jira/browse/KAFKA-13433 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Affects Versions: 3.0.0 >Reporter: Aiden Gong >Priority: Blocker > Attachments: image-2021-11-04-16-25-18-890.png, > image-2021-11-04-16-25-18-975.png > > > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. Please refer attachments for > more detail. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-13433) JsonConverter's method convertToJson when field is optional with default value and value is null, return default value.
[ https://issues.apache.org/jira/browse/KAFKA-13433?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17438553#comment-17438553 ] Aiden Gong commented on KAFKA-13433: I will submit a pr to fix this issue. Please assignee to me. Thanks! > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. > > > Key: KAFKA-13433 > URL: https://issues.apache.org/jira/browse/KAFKA-13433 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Affects Versions: 3.0.0 >Reporter: Aiden Gong >Priority: Blocker > Attachments: image-2021-11-04-16-25-18-890.png, > image-2021-11-04-16-25-18-975.png > > > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. Please refer attachments for > more detail. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-13433) JsonConverter's method convertToJson when field is optional with default value and value is null, return default value.
[ https://issues.apache.org/jira/browse/KAFKA-13433?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17438556#comment-17438556 ] Aiden Gong commented on KAFKA-13433: Class org.apache.kafka.connect.json.JsonConverter line 704, should use 'struct.getWithoutDefault(field)' instead of 'struct.get(field)' . > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. > > > Key: KAFKA-13433 > URL: https://issues.apache.org/jira/browse/KAFKA-13433 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Affects Versions: 3.0.0 >Reporter: Aiden Gong >Priority: Blocker > Attachments: image-2021-11-04-16-25-18-890.png, > image-2021-11-04-16-25-18-975.png > > > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. Please refer attachments for > more detail. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-13433) JsonConverter's method convertToJson when field is optional with default value and value is null, return default value.
[ https://issues.apache.org/jira/browse/KAFKA-13433?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Aiden Gong updated KAFKA-13433: --- Priority: Minor (was: Blocker) > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. > > > Key: KAFKA-13433 > URL: https://issues.apache.org/jira/browse/KAFKA-13433 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Affects Versions: 3.0.0 >Reporter: Aiden Gong >Priority: Minor > Attachments: image-2021-11-04-16-25-18-890.png, > image-2021-11-04-16-25-18-975.png > > > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. Please refer attachments for > more detail. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-13433) JsonConverter's method convertToJson when field is optional with default value and value is null, return default value.
[ https://issues.apache.org/jira/browse/KAFKA-13433?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Aiden Gong updated KAFKA-13433: --- Priority: Major (was: Minor) > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. > > > Key: KAFKA-13433 > URL: https://issues.apache.org/jira/browse/KAFKA-13433 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Affects Versions: 3.0.0 >Reporter: Aiden Gong >Priority: Major > Attachments: image-2021-11-04-16-25-18-890.png, > image-2021-11-04-16-25-18-975.png > > > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. Please refer attachments for > more detail. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-13433) JsonConverter's method convertToJson when field is optional with default value and value is null, return default value.
[ https://issues.apache.org/jira/browse/KAFKA-13433?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Aiden Gong updated KAFKA-13433: --- Labels: connect-api (was: ) > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. > > > Key: KAFKA-13433 > URL: https://issues.apache.org/jira/browse/KAFKA-13433 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Affects Versions: 3.0.0 >Reporter: Aiden Gong >Priority: Major > Labels: connect-api > Attachments: image-2021-11-04-16-25-18-890.png, > image-2021-11-04-16-25-18-975.png > > > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. Please refer attachments for > more detail. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (KAFKA-13433) JsonConverter's method convertToJson when field is optional with default value and value is null, return default value.
[ https://issues.apache.org/jira/browse/KAFKA-13433?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Aiden Gong updated KAFKA-13433: --- Labels: connect (was: connect-api) > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. > > > Key: KAFKA-13433 > URL: https://issues.apache.org/jira/browse/KAFKA-13433 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Affects Versions: 3.0.0 >Reporter: Aiden Gong >Priority: Major > Labels: connect > Attachments: image-2021-11-04-16-25-18-890.png, > image-2021-11-04-16-25-18-975.png > > > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. Please refer attachments for > more detail. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [kafka] dajac merged pull request #11463: KAFKA-13430: Remove broker-wide quota properties from the documentation
dajac merged pull request #11463: URL: https://github.com/apache/kafka/pull/11463 -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #11463: KAFKA-13430: Remove broker-wide quota properties from the documentation
dajac commented on pull request #11463: URL: https://github.com/apache/kafka/pull/11463#issuecomment-960571438 Merged to trunk and 3.1. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Commented] (KAFKA-13351) Add possibility to write kafka headers in Kafka Console Producer
[ https://issues.apache.org/jira/browse/KAFKA-13351?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17438573#comment-17438573 ] Seweryn Habdank-Wojewodzki commented on KAFKA-13351: Thanks a lot! > Add possibility to write kafka headers in Kafka Console Producer > > > Key: KAFKA-13351 > URL: https://issues.apache.org/jira/browse/KAFKA-13351 > Project: Kafka > Issue Type: Wish > Components: tools >Affects Versions: 2.8.1 >Reporter: Seweryn Habdank-Wojewodzki >Assignee: Florin Akermann >Priority: Major > > Dears, > Currently there is an asymetry between Kafka Console Consumer and Producer. > Kafka Consumer can display headers (KAFKA-6733), but Kafka Producer cannot > produce them. > It would be good to unify this and add possibility to Kafka Console Producer > to produce them. > Similar ticket is here: KAFKA-6574, but it is very old and does not > represents current state of the software. > Please consider this. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (KAFKA-13430) Remove broker-wide quota properties from the documentation
[ https://issues.apache.org/jira/browse/KAFKA-13430?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17438574#comment-17438574 ] ASF GitHub Bot commented on KAFKA-13430: dajac merged pull request #380: URL: https://github.com/apache/kafka-site/pull/380 -- 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: dev-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Remove broker-wide quota properties from the documentation > -- > > Key: KAFKA-13430 > URL: https://issues.apache.org/jira/browse/KAFKA-13430 > Project: Kafka > Issue Type: Bug > Components: documentation >Affects Versions: 3.0.0 >Reporter: Dongjin Lee >Assignee: Dongjin Lee >Priority: Major > > I found this problem while working on > [KAFKA-13341|https://issues.apache.org/jira/browse/KAFKA-13341]. > Broker-wide quota properties ({{quota.producer.default}}, > {{quota.consumer.default}}) are [removed in > 3.0|https://issues.apache.org/jira/browse/KAFKA-12591], but it is not applied > to the documentation yet. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Resolved] (KAFKA-13430) Remove broker-wide quota properties from the documentation
[ https://issues.apache.org/jira/browse/KAFKA-13430?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] David Jacot resolved KAFKA-13430. - Fix Version/s: 3.0.0 3.1.0 Reviewer: David Jacot Resolution: Fixed > Remove broker-wide quota properties from the documentation > -- > > Key: KAFKA-13430 > URL: https://issues.apache.org/jira/browse/KAFKA-13430 > Project: Kafka > Issue Type: Bug > Components: documentation >Affects Versions: 3.0.0 >Reporter: Dongjin Lee >Assignee: Dongjin Lee >Priority: Major > Fix For: 3.1.0, 3.0.0 > > > I found this problem while working on > [KAFKA-13341|https://issues.apache.org/jira/browse/KAFKA-13341]. > Broker-wide quota properties ({{quota.producer.default}}, > {{quota.consumer.default}}) are [removed in > 3.0|https://issues.apache.org/jira/browse/KAFKA-12591], but it is not applied > to the documentation yet. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [kafka] showuon commented on pull request #11429: KAFKA-13396: allow create topic without partition/replicaFactor
showuon commented on pull request #11429: URL: https://github.com/apache/kafka/pull/11429#issuecomment-960575313 @ijuma , please help check it when available. This is a regression issue, so I think we should put it v3.1.0. Thank you. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Commented] (KAFKA-12487) Sink connectors do not work with the cooperative consumer rebalance protocol
[ https://issues.apache.org/jira/browse/KAFKA-12487?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17438579#comment-17438579 ] David Jacot commented on KAFKA-12487: - Hi [~kkonstantine]. My understanding of the issue is that we are fixing a bug here. Therefore, merging it to 3.1 and 3.0 is fine for me. Please correct if it is not a bug. > Sink connectors do not work with the cooperative consumer rebalance protocol > > > Key: KAFKA-12487 > URL: https://issues.apache.org/jira/browse/KAFKA-12487 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Affects Versions: 2.4.2, 2.5.2, 2.8.0, 2.7.1, 2.6.2, 3.0.0 >Reporter: Chris Egerton >Assignee: Chris Egerton >Priority: Blocker > Fix For: 3.1.0 > > > The {{ConsumerRebalanceListener}} used by the framework to respond to > rebalance events in consumer groups for sink tasks is hard-coded with the > assumption that the consumer performs rebalances eagerly. In other words, it > assumes that whenever {{onPartitionsRevoked}} is called, all partitions have > been revoked from that consumer, and whenever {{onPartitionsAssigned}} is > called, the partitions passed in to that method comprise the complete set of > topic partitions assigned to that consumer. > See the [WorkerSinkTask.HandleRebalance > class|https://github.com/apache/kafka/blob/b96fc7892f1e885239d3290cf509e1d1bb41e7db/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java#L669-L730] > for the specifics. > > One issue this can cause is silently ignoring to-be-committed offsets > provided by sink tasks, since the framework ignores offsets provided by tasks > in their {{preCommit}} method if it does not believe that the consumer for > that task is currently assigned the topic partition for that offset. See > these lines in the [WorkerSinkTask::commitOffsets > method|https://github.com/apache/kafka/blob/b96fc7892f1e885239d3290cf509e1d1bb41e7db/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java#L429-L430] > for reference. > > This may not be the only issue caused by configuring a sink connector's > consumer to use cooperative rebalancing. Rigorous unit and integration > testing should be added before claiming that the Connect framework supports > the use of cooperative consumers with sink connectors. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [kafka] mimaison commented on a change in pull request #11401: KAFKA-13255: use exclude filter for new topics
mimaison commented on a change in pull request #11401: URL: https://github.com/apache/kafka/pull/11401#discussion_r742653123 ## File path: connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConnectorTest.java ## @@ -152,6 +155,48 @@ public void testConfigPropertyFiltering() { .anyMatch(x -> x.name().equals("min.insync.replicas")), "should not replicate excluded properties"); } +@Test +public void testNewTopicConfigs() throws Exception { +Map filterConfig = new HashMap<>(); + filterConfig.put(DefaultConfigPropertyFilter.CONFIG_PROPERTIES_EXCLUDE_CONFIG, "follower\\.replication\\.throttled\\.replicas, " ++ "leader\\.replication\\.throttled\\.replicas, " ++ "message\\.timestamp\\.difference\\.max\\.ms, " ++ "message\\.timestamp\\.type, " ++ "unclean\\.leader\\.election\\.enable, " ++ "min\\.insync\\.replicas," ++ "exclude_param.*"); +DefaultConfigPropertyFilter filter = new DefaultConfigPropertyFilter(); +filter.configure(filterConfig); + +MirrorSourceConnector connector = spy(new MirrorSourceConnector(new SourceAndTarget("source", "target"), +new DefaultReplicationPolicy(), x -> true, filter)); + +final String topic = "testtopic"; +List entries = new ArrayList<>(); +entries.add(new ConfigEntry("name-1", "value-1")); +entries.add(new ConfigEntry("exclude_param.param1", "value-param1")); +entries.add(new ConfigEntry("min.insync.replicas", "2")); +Config config = new Config(entries); +doReturn(Collections.singletonMap(topic, config)).when(connector).describeTopicConfigs(any()); +doAnswer(invocation -> { +Map newTopics = invocation.getArgument(0); +assertNotNull(newTopics.get("source." + topic)); +Map targetConfig = newTopics.get("source." + topic).configs(); + +// property 'name-1' isn't defined in the exclude filter -> should be replicated +assertNotNull(targetConfig.get("name-1"), "should replicate properties"); + +// this property is in default list, just double check it: +String prop1 = "min.insync.replicas"; +assertNull(targetConfig.get(prop1), "should not replicate excluded properties " + prop1); +// this property is only in exclude filter custom parameter, also tests regex on the way: +String prop2 = "exclude_param.param1"; +assertNull(targetConfig.get(prop2), "should not replicate excluded properties " + prop2); +return null; +}).when(connector).createNewTopics(any()); +connector.createNewTopics(Collections.singleton(topic), Collections.singletonMap(topic, 1L)); +} Review comment: I agree that calling verify on `targetConfig()` is not the best. But I think we should call verify on `createNewTopics(any())` to ensure our mocked method, and the assertions, actually run. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Comment Edited] (KAFKA-13433) JsonConverter's method convertToJson when field is optional with default value and value is null, return default value.
[ https://issues.apache.org/jira/browse/KAFKA-13433?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17438556#comment-17438556 ] Aiden Gong edited comment on KAFKA-13433 at 11/4/21, 9:38 AM: -- Class org.apache.kafka.connect.json.JsonConverter line 704, should use 'struct.getWithoutDefault(field)' instead of 'struct.get(field.name())' . was (Author: aiden gong): Class org.apache.kafka.connect.json.JsonConverter line 704, should use 'struct.getWithoutDefault(field)' instead of 'struct.get(field)' . > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. > > > Key: KAFKA-13433 > URL: https://issues.apache.org/jira/browse/KAFKA-13433 > Project: Kafka > Issue Type: Bug > Components: KafkaConnect >Affects Versions: 3.0.0 >Reporter: Aiden Gong >Priority: Major > Labels: connect > Attachments: image-2021-11-04-16-25-18-890.png, > image-2021-11-04-16-25-18-975.png > > > JsonConverter's method convertToJson when field is optional with default > value and value is null, return default value. Please refer attachments for > more detail. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [kafka] Gerrrr commented on pull request #11447: KAFKA-13024: Use not-null filter only in optimizable repartitions
Ge commented on pull request #11447: URL: https://github.com/apache/kafka/pull/11447#issuecomment-960765623 Thanks for the review! I agree that solving this issue by removing the not-null optimization completely is very much suboptimal. 8f176cb reverts the first commit. 7c6e97f introduces another approach where `UnoptimizableRepartitionNode` adds the not-null filter iff all downstream operations drop the null keys. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] showuon opened a new pull request #11467: MINOR: fix java doc in kafkaProducer
showuon opened a new pull request #11467: URL: https://github.com/apache/kafka/pull/11467 *More detailed description of your change, if necessary. The PR title and PR message become the squashed commit message, so use a separate comment to ping reviewers.* *Summary of testing strategy (including rationale) for the feature or bug fix. Unit and/or integration tests are expected for any behaviour change and system tests should be considered for larger changes.* ### Committer Checklist (excluded from commit message) - [ ] Verify design and implementation - [ ] Verify test coverage and CI build status - [ ] Verify documentation (including upgrade notes) -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11467: MINOR: fix java doc in kafkaProducer
showuon commented on a change in pull request #11467: URL: https://github.com/apache/kafka/pull/11467#discussion_r742807104 ## File path: clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java ## @@ -117,14 +115,15 @@ * as well as a background I/O thread that is responsible for turning these records into requests and transmitting them * to the cluster. Failure to close the producer after use will leak these resources. * - * The {@link #send(ProducerRecord) send()} method is asynchronous. When called it adds the record to a buffer of pending record sends + * The {@link #send(ProducerRecord) send()} method is asynchronous. When called, it adds the record to a buffer of pending sending record * and immediately returns. This allows the producer to batch together individual records for efficiency. * * The acks config controls the criteria under which requests are considered complete. The "all" setting * we have specified will result in blocking on the full commit of the record, the slowest but most durable setting. * - * If the request fails, the producer can automatically retry, though since we have specified retries - * as 0 it won't. Enabling retries also opens up the possibility of duplicates (see the documentation on + * If the request fails, the producer can automatically retry. We have specified retries default as Integer.MAX_VALUE. + * It's recommended to use delivery.timeout.ms to control retry behavior, instead of retries. + * Enabling retries also opens up the possibility of duplicates (see the documentation on Review comment: After [KIP-91](https://cwiki.apache.org/confluence/display/KAFKA/KIP-91+Provide+Intuitive+User+Timeouts+in+The+Producer), we set `retries` as max value, and it's recommended to use `delivery.timeout.ms` to control retry behavior. Update it. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11467: MINOR: fix java doc in kafkaProducer
showuon commented on a change in pull request #11467: URL: https://github.com/apache/kafka/pull/11467#discussion_r742808430 ## File path: clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java ## @@ -100,8 +100,6 @@ * {@code * Properties props = new Properties(); * props.put("bootstrap.servers", "localhost:9092"); - * props.put("acks", "all"); - * props.put("retries", 0); Review comment: In V3.0, `acks` is default to `all`, and `retires` is not recommended to set to 0 after KIP-91(https://cwiki.apache.org/confluence/display/KAFKA/KIP-91+Provide+Intuitive+User+Timeouts+in+The+Producer) -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11467: MINOR: fix java doc in kafkaProducer
showuon commented on a change in pull request #11467: URL: https://github.com/apache/kafka/pull/11467#discussion_r742808430 ## File path: clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java ## @@ -100,8 +100,6 @@ * {@code * Properties props = new Properties(); * props.put("bootstrap.servers", "localhost:9092"); - * props.put("acks", "all"); - * props.put("retries", 0); Review comment: In V3.0, `acks` is default to `all`, and `retires` is not recommended to set to 0 after [KIP-91](https://cwiki.apache.org/confluence/display/KAFKA/KIP-91+Provide+Intuitive+User+Timeouts+in+The+Producer) . -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #11467: MINOR: fix java doc in kafkaProducer
showuon commented on pull request #11467: URL: https://github.com/apache/kafka/pull/11467#issuecomment-960867632 @dajac , please help review this java doc update, and I think this fix should put into V3.1.0. WDYT? -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #11429: KAFKA-13396: allow create topic without partition/replicaFactor
ijuma commented on pull request #11429: URL: https://github.com/apache/kafka/pull/11429#issuecomment-960945328 @showuon we should have this in 3.0.1 too, right? -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #11429: KAFKA-13396: allow create topic without partition/replicaFactor
showuon commented on pull request #11429: URL: https://github.com/apache/kafka/pull/11429#issuecomment-960951629 Yes, cherry-pick back to v3.0.1, too, please. Thank you! -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] ijuma merged pull request #11429: KAFKA-13396: allow create topic without partition/replicaFactor
ijuma merged pull request #11429: URL: https://github.com/apache/kafka/pull/11429 -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
dajac commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r742933244 ## File path: clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java ## @@ -428,82 +442,165 @@ public void testIdUsageRevokedOnIdDowngrade() { List partitions = Arrays.asList(0, 1); partitions.forEach(partition -> { String testType = partition == 0 ? "updating a partition" : "adding a new partition"; -Map topicIds = Collections.singletonMap("foo", Uuid.randomUuid()); +Uuid fooId = Uuid.randomUuid(); FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); FetchSessionHandler.Builder builder = handler.newBuilder(); -builder.add(new TopicPartition("foo", 0), topicIds.get("foo"), -new FetchRequest.PartitionData(0, 100, 200, Optional.empty())); +builder.add(new TopicPartition("foo", 0), +new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty())); FetchSessionHandler.FetchRequestData data = builder.build(); -assertMapsEqual(reqMap(new ReqEntry("foo", 0, 0, 100, 200)), +assertMapsEqual(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200)), data.toSend(), data.sessionPartitions()); assertTrue(data.metadata().isFull()); assertTrue(data.canUseTopicIds()); FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, -respMap(new RespEntry("foo", 0, 10, 20)), topicIds); +respMap(new RespEntry("foo", 0, fooId, 10, 20))); handler.handleResponse(resp, ApiKeys.FETCH.latestVersion()); // Try to remove a topic ID from an existing topic partition (0) or add a new topic partition (1) without an ID. FetchSessionHandler.Builder builder2 = handler.newBuilder(); -builder2.add(new TopicPartition("foo", partition), Uuid.ZERO_UUID, -new FetchRequest.PartitionData(10, 110, 210, Optional.empty())); +builder2.add(new TopicPartition("foo", partition), +new FetchRequest.PartitionData(Uuid.ZERO_UUID, 10, 110, 210, Optional.empty())); FetchSessionHandler.FetchRequestData data2 = builder2.build(); -// Should have the same session ID and next epoch, but can no longer use topic IDs. -// The receiving broker will close the session if we were previously using topic IDs. +// Should have the same session ID, and next epoch and can no longer use topic IDs. +// The receiving broker will handle closing the session. assertEquals(123, data2.metadata().sessionId(), "Did not use same session when " + testType); assertEquals(1, data2.metadata().epoch(), "Did not have correct epoch when " + testType); assertFalse(data2.canUseTopicIds()); }); } +@ParameterizedTest +@ValueSource(booleans = {true, false}) +public void testTopicIdReplaced(boolean fetchRequestUsesIds) { +TopicPartition tp = new TopicPartition("foo", 0); +FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); +FetchSessionHandler.Builder builder = handler.newBuilder(); +Uuid topicId1 = Uuid.randomUuid(); +builder.add(tp, +new FetchRequest.PartitionData(topicId1, 0, 100, 200, Optional.empty())); +FetchSessionHandler.FetchRequestData data = builder.build(); +assertMapsEqual(reqMap(new ReqEntry("foo", topicId1, 0, 0, 100, 200)), +data.toSend(), data.sessionPartitions()); +assertTrue(data.metadata().isFull()); +assertTrue(data.canUseTopicIds()); + +FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, +respMap(new RespEntry("foo", 0, Uuid.ZERO_UUID, 10, 20))); +handler.handleResponse(resp, (short) 12); + +// Try to add a new topic ID. +FetchSessionHandler.Builder builder2 = handler.newBuilder(); +Uuid topicId2 = Uuid.randomUuid(); +// Use the same data besides the topic ID. +FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId2, 0, 100, 200, Optional.empty()); +builder2.add(tp, partitionData); +FetchSessionHandler.FetchRequestData data2 = builder2.build(); +// The old topic ID partition should be in toReplace, and the new one should be in toSend. +assertMapsEqual(reqMap(new ReqEntry("foo", topicId2, 0, 0, 100, 200)), +data2.toSend(), data2.sessionPartitions()); +assertEquals(Collections.singletonList(new TopicIdPartition(topicId1, tp)), data2.toReplace()); +// Should have the same session ID, and next epoch and can use topic IDs. +assertEquals(123, data2.metadata().sessionId(), "Did not use same s
[GitHub] [kafka] dajac commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
dajac commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743011890 ## File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala ## @@ -1361,102 +1542,113 @@ class FetchSessionTest { val resp4 = context2.updateAndGenerateResponseData(respData) assertEquals(Errors.NONE, resp4.error) assertEquals(resp1.sessionId, resp4.sessionId) -assertEquals(Utils.mkSet(tp1, tp2), resp4.responseData(topicNames, request2.version).keySet) +assertEquals(Utils.mkSet(tp1.topicPartition, tp2.topicPartition), resp4.responseData(topicNames, request2.version).keySet) } @Test def testDeprioritizesPartitionsWithRecordsOnly(): Unit = { val time = new MockTime() val cache = new FetchSessionCache(10, 1000) val fetchManager = new FetchManager(time, cache) -val tp1 = new TopicPartition("foo", 1) -val tp2 = new TopicPartition("bar", 2) -val tp3 = new TopicPartition("zar", 3) val topicIds = Map("foo" -> Uuid.randomUuid(), "bar" -> Uuid.randomUuid(), "zar" -> Uuid.randomUuid()).asJava val topicNames = topicIds.asScala.map(_.swap).asJava +val tp1 = new TopicIdPartition(topicIds.get("foo"), new TopicPartition("foo", 1)) +val tp2 = new TopicIdPartition(topicIds.get("bar"), new TopicPartition("bar", 2)) +val tp3 = new TopicIdPartition(topicIds.get("zar"), new TopicPartition("zar", 3)) -val reqData = new util.LinkedHashMap[TopicPartition, FetchRequest.PartitionData] -reqData.put(tp1, new FetchRequest.PartitionData(100, 0, 1000, Optional.of(5), Optional.of(4))) -reqData.put(tp2, new FetchRequest.PartitionData(100, 0, 1000, Optional.of(5), Optional.of(4))) -reqData.put(tp3, new FetchRequest.PartitionData(100, 0, 1000, Optional.of(5), Optional.of(4))) +val reqData = new util.LinkedHashMap[TopicIdPartition, FetchRequest.PartitionData] +reqData.put(tp1, new FetchRequest.PartitionData(topicIds.get("foo"), 100, 0, 1000, Optional.of(5), Optional.of(4))) +reqData.put(tp2, new FetchRequest.PartitionData(topicIds.get("bar"), 100, 0, 1000, Optional.of(5), Optional.of(4))) +reqData.put(tp3, new FetchRequest.PartitionData(topicIds.get("zar"), 100, 0, 1000, Optional.of(5), Optional.of(4))) // Full fetch context returns all partitions in the response val context1 = fetchManager.newContext(ApiKeys.FETCH.latestVersion(), JFetchMetadata.INITIAL, false, - reqData, Collections.emptyList(), topicIds) + reqData, Collections.emptyList(), topicNames) assertEquals(classOf[FullFetchContext], context1.getClass) -val respData1 = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData] +val respData1 = new util.LinkedHashMap[TopicIdPartition, FetchResponseData.PartitionData] respData1.put(tp1, new FetchResponseData.PartitionData() - .setPartitionIndex(tp1.partition) + .setPartitionIndex(tp1.topicPartition.partition) .setHighWatermark(50) .setLastStableOffset(50) .setLogStartOffset(0)) respData1.put(tp2, new FetchResponseData.PartitionData() - .setPartitionIndex(tp2.partition) + .setPartitionIndex(tp2.topicPartition.partition) .setHighWatermark(50) .setLastStableOffset(50) .setLogStartOffset(0)) respData1.put(tp3, new FetchResponseData.PartitionData() - .setPartitionIndex(tp3.partition) + .setPartitionIndex(tp3.topicPartition.partition) .setHighWatermark(50) .setLastStableOffset(50) .setLogStartOffset(0)) val resp1 = context1.updateAndGenerateResponseData(respData1) assertEquals(Errors.NONE, resp1.error) assertNotEquals(INVALID_SESSION_ID, resp1.sessionId) -assertEquals(Utils.mkSet(tp1, tp2, tp3), resp1.responseData(topicNames, ApiKeys.FETCH.latestVersion()).keySet()) +assertEquals(Utils.mkSet(tp1.topicPartition, tp2.topicPartition, tp3.topicPartition), resp1.responseData(topicNames, ApiKeys.FETCH.latestVersion()).keySet()) // Incremental fetch context returns partitions with changes but only deprioritizes // the partitions with records val context2 = fetchManager.newContext(ApiKeys.FETCH.latestVersion(), new JFetchMetadata(resp1.sessionId, 1), false, - reqData, Collections.emptyList(), topicIds) + reqData, Collections.emptyList(), topicNames) assertEquals(classOf[IncrementalFetchContext], context2.getClass) // Partitions are ordered in the session as per last response assertPartitionsOrder(context2, Seq(tp1, tp2, tp3)) // Response is empty -val respData2 = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData] +val respData2 = new util.LinkedHashMap[TopicIdPartition, FetchResponseData.PartitionData] val resp2 = context2.updateAndGenerateResponseData(respData2) assertEquals(Errors.NONE, resp2.error) assertEquals(resp1.sessionId, resp2.sessionId) assertEquals(Collections.emptySe
[GitHub] [kafka] dajac commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
dajac commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743041695 ## File path: clients/src/main/java/org/apache/kafka/clients/FetchSessionHandler.java ## @@ -285,52 +268,57 @@ public FetchRequestData build() { if (nextMetadata.isFull()) { if (log.isDebugEnabled()) { log.debug("Built full fetch {} for node {} with {}.", - nextMetadata, node, partitionsToLogString(next.keySet())); +nextMetadata, node, topicPartitionsToLogString(next.keySet())); } sessionPartitions = next; next = null; +Map toSend = +Collections.unmodifiableMap(new LinkedHashMap<>(sessionPartitions)); // Only add topic IDs to the session if we are using topic IDs. if (canUseTopicIds) { -sessionTopicIds = topicIds; -sessionTopicNames = new HashMap<>(topicIds.size()); -topicIds.forEach((name, id) -> sessionTopicNames.put(id, name)); +Map> newTopicNames = sessionPartitions.entrySet().stream().collect(Collectors.groupingByConcurrent(entry -> entry.getValue().topicId, +Collectors.mapping(entry -> entry.getKey().topic(), Collectors.toSet(; Review comment: I think that the grouping is slower because it has to allocate another Map, Sets for each Uuid, etc. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
dajac commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743044369 ## File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala ## @@ -374,7 +374,7 @@ abstract class AbstractFetcherThread(name: String, } } } catch { -case ime@( _: CorruptRecordException | _: InvalidRecordException) => +case ime@(_: CorruptRecordException | _: InvalidRecordException) => Review comment: I think that would for instance append when the controller fails over to an older IBP during an upgrade. This should remove the topic ids which means that v12 will be used for the next fetch request and trigger a FETCH_SESSION_TOPIC_ID_ERROR. In this particular case, re-trying directly would be the optimal way to proceed for a follower. I wonder if they are other cases to consider here. For the consumer, it is definitely different. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
dajac commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743044970 ## File path: core/src/main/scala/kafka/server/FetchSession.scala ## @@ -163,18 +173,35 @@ class CachedPartition(val topic: String, mustRespond } - override def hashCode: Int = Objects.hash(new TopicPartition(topic, partition), topicId) + /** + * We have different equality checks depending on whether topic IDs are used. + * This means we need a different hash function as well. We use name to calculate the hash if the ID is zero and unused. + * Otherwise, we use the topic ID in the hash calculation. + * + * @return the hash code for the CachedPartition depending on what request version we are using. + */ + override def hashCode: Int = if (topicId != Uuid.ZERO_UUID) (31 * partition) + topicId.hashCode else +(31 * partition) + topic.hashCode def canEqual(that: Any): Boolean = that.isInstanceOf[CachedPartition] + /** + * We have different equality checks depending on whether topic IDs are used. + * + * This is because when we use topic IDs, a partition with a given ID and an unknown name is the same as a partition with that + * ID and a known name. This means we can only use topic ID and partition when determining equality. + * + * On the other hand, if we are using topic names, all IDs are zero. This means we can only use topic name and partition + * when determining equality. + */ override def equals(that: Any): Boolean = that match { case that: CachedPartition => this.eq(that) || (that.canEqual(this) && Review comment: Right. It seems to be that the `canEqual(this)` does not make any sense here. Could you double check? -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
dajac commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743046003 ## File path: core/src/main/scala/kafka/server/KafkaApis.scala ## @@ -708,40 +701,41 @@ class KafkaApis(val requestChannel: RequestChannel, None } -val erroneous = mutable.ArrayBuffer[(TopicPartition, FetchResponseData.PartitionData)]() -val interesting = mutable.ArrayBuffer[(TopicPartition, FetchRequest.PartitionData)]() -val sessionTopicIds = mutable.Map[String, Uuid]() +val erroneous = mutable.ArrayBuffer[(TopicIdPartition, FetchResponseData.PartitionData)]() +val interesting = mutable.ArrayBuffer[(TopicIdPartition, FetchRequest.PartitionData)]() if (fetchRequest.isFromFollower) { // The follower must have ClusterAction on ClusterResource in order to fetch partition data. if (authHelper.authorize(request.context, CLUSTER_ACTION, CLUSTER, CLUSTER_NAME)) { -fetchContext.foreachPartition { (topicPartition, topicId, data) => - sessionTopicIds.put(topicPartition.topic(), topicId) - if (!metadataCache.contains(topicPartition)) -erroneous += topicPartition -> FetchResponse.partitionResponse(topicPartition.partition, Errors.UNKNOWN_TOPIC_OR_PARTITION) +fetchContext.foreachPartition { (topicIdPartition, data) => + if (topicIdPartition.topicPartition.topic == null ) +erroneous += topicIdPartition -> FetchResponse.partitionResponse(topicIdPartition, Errors.UNKNOWN_TOPIC_ID) + else if (!metadataCache.contains(topicIdPartition.topicPartition)) +erroneous += topicIdPartition -> FetchResponse.partitionResponse(topicIdPartition, Errors.UNKNOWN_TOPIC_OR_PARTITION) else -interesting += (topicPartition -> data) +interesting += (topicIdPartition -> data) } } else { -fetchContext.foreachPartition { (part, topicId, _) => - sessionTopicIds.put(part.topic(), topicId) - erroneous += part -> FetchResponse.partitionResponse(part.partition, Errors.TOPIC_AUTHORIZATION_FAILED) +fetchContext.foreachPartition { (topicIdPartition, _) => + erroneous += topicIdPartition -> FetchResponse.partitionResponse(topicIdPartition, Errors.TOPIC_AUTHORIZATION_FAILED) Review comment: I guess that it does not change much in the end. I was considering this in order to be consistent with how we handle this for the consumer. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
dajac commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743046927 ## File path: clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java ## @@ -66,16 +69,28 @@ public PartitionData( int maxBytes, Optional currentLeaderEpoch ) { -this(fetchOffset, logStartOffset, maxBytes, currentLeaderEpoch, Optional.empty()); +this(Uuid.ZERO_UUID, fetchOffset, logStartOffset, maxBytes, currentLeaderEpoch, Optional.empty()); Review comment: Yeah, that's a good question. I guess that that constructor is convenient for tests but might be bug prone in the regular code. I am tempted to remove it entirely What do you think? -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
dajac commented on pull request #11331: URL: https://github.com/apache/kafka/pull/11331#issuecomment-961259072 @jolshan It seems that there are a few compilation errors, at least for `JDK 8 and Scala 2.12`. Could you check? -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] bdesert commented on a change in pull request #11401: KAFKA-13255: use exclude filter for new topics
bdesert commented on a change in pull request #11401: URL: https://github.com/apache/kafka/pull/11401#discussion_r743062367 ## File path: connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConnectorTest.java ## @@ -152,6 +155,48 @@ public void testConfigPropertyFiltering() { .anyMatch(x -> x.name().equals("min.insync.replicas")), "should not replicate excluded properties"); } +@Test +public void testNewTopicConfigs() throws Exception { +Map filterConfig = new HashMap<>(); + filterConfig.put(DefaultConfigPropertyFilter.CONFIG_PROPERTIES_EXCLUDE_CONFIG, "follower\\.replication\\.throttled\\.replicas, " ++ "leader\\.replication\\.throttled\\.replicas, " ++ "message\\.timestamp\\.difference\\.max\\.ms, " ++ "message\\.timestamp\\.type, " ++ "unclean\\.leader\\.election\\.enable, " ++ "min\\.insync\\.replicas," ++ "exclude_param.*"); +DefaultConfigPropertyFilter filter = new DefaultConfigPropertyFilter(); +filter.configure(filterConfig); + +MirrorSourceConnector connector = spy(new MirrorSourceConnector(new SourceAndTarget("source", "target"), +new DefaultReplicationPolicy(), x -> true, filter)); + +final String topic = "testtopic"; +List entries = new ArrayList<>(); +entries.add(new ConfigEntry("name-1", "value-1")); +entries.add(new ConfigEntry("exclude_param.param1", "value-param1")); +entries.add(new ConfigEntry("min.insync.replicas", "2")); +Config config = new Config(entries); +doReturn(Collections.singletonMap(topic, config)).when(connector).describeTopicConfigs(any()); +doAnswer(invocation -> { +Map newTopics = invocation.getArgument(0); +assertNotNull(newTopics.get("source." + topic)); +Map targetConfig = newTopics.get("source." + topic).configs(); + +// property 'name-1' isn't defined in the exclude filter -> should be replicated +assertNotNull(targetConfig.get("name-1"), "should replicate properties"); + +// this property is in default list, just double check it: +String prop1 = "min.insync.replicas"; +assertNull(targetConfig.get(prop1), "should not replicate excluded properties " + prop1); +// this property is only in exclude filter custom parameter, also tests regex on the way: +String prop2 = "exclude_param.param1"; +assertNull(targetConfig.get(prop2), "should not replicate excluded properties " + prop2); +return null; +}).when(connector).createNewTopics(any()); +connector.createNewTopics(Collections.singleton(topic), Collections.singletonMap(topic, 1L)); +} Review comment: agree. with a small correction of `createNewTopics(any(), any())`. since the `targetConfig()` is being called from overloaded method with two params. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] dajac opened a new pull request #11468: WIP
dajac opened a new pull request #11468: URL: https://github.com/apache/kafka/pull/11468 *More detailed description of your change, if necessary. The PR title and PR message become the squashed commit message, so use a separate comment to ping reviewers.* *Summary of testing strategy (including rationale) for the feature or bug fix. Unit and/or integration tests are expected for any behaviour change and system tests should be considered for larger changes.* ### Committer Checklist (excluded from commit message) - [ ] Verify design and implementation - [ ] Verify test coverage and CI build status - [ ] Verify documentation (including upgrade notes) -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #11429: KAFKA-13396: allow create topic without partition/replicaFactor
dajac commented on pull request #11429: URL: https://github.com/apache/kafka/pull/11429#issuecomment-961343546 @ijuma Should we pick this one to the 3.1 branch as well? -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #11467: MINOR: fix java doc in kafkaProducer
dajac commented on pull request #11467: URL: https://github.com/apache/kafka/pull/11467#issuecomment-961343975 @showuon Thanks. I will take a look soon. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #11429: KAFKA-13396: allow create topic without partition/replicaFactor
ijuma commented on pull request #11429: URL: https://github.com/apache/kafka/pull/11429#issuecomment-961367197 @dajac Makes sense, pushed to trunk, 3.1 and 3.0. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743239514 ## File path: clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java ## @@ -202,29 +203,30 @@ public void testSessionless() { FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); FetchSessionHandler.Builder builder = handler.newBuilder(); addTopicId(topicIds, topicNames, "foo", version); -builder.add(new TopicPartition("foo", 0), topicIds.getOrDefault("foo", Uuid.ZERO_UUID), -new FetchRequest.PartitionData(0, 100, 200, Optional.empty())); -builder.add(new TopicPartition("foo", 1), topicIds.getOrDefault("foo", Uuid.ZERO_UUID), -new FetchRequest.PartitionData(10, 110, 210, Optional.empty())); +Uuid fooId = topicIds.getOrDefault("foo", Uuid.ZERO_UUID); +builder.add(new TopicPartition("foo", 0), +new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty())); Review comment: I moved some back. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743240074 ## File path: clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java ## @@ -428,82 +442,165 @@ public void testIdUsageRevokedOnIdDowngrade() { List partitions = Arrays.asList(0, 1); partitions.forEach(partition -> { String testType = partition == 0 ? "updating a partition" : "adding a new partition"; -Map topicIds = Collections.singletonMap("foo", Uuid.randomUuid()); +Uuid fooId = Uuid.randomUuid(); FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); FetchSessionHandler.Builder builder = handler.newBuilder(); -builder.add(new TopicPartition("foo", 0), topicIds.get("foo"), -new FetchRequest.PartitionData(0, 100, 200, Optional.empty())); +builder.add(new TopicPartition("foo", 0), +new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty())); FetchSessionHandler.FetchRequestData data = builder.build(); -assertMapsEqual(reqMap(new ReqEntry("foo", 0, 0, 100, 200)), +assertMapsEqual(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200)), data.toSend(), data.sessionPartitions()); assertTrue(data.metadata().isFull()); assertTrue(data.canUseTopicIds()); FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, -respMap(new RespEntry("foo", 0, 10, 20)), topicIds); +respMap(new RespEntry("foo", 0, fooId, 10, 20))); handler.handleResponse(resp, ApiKeys.FETCH.latestVersion()); // Try to remove a topic ID from an existing topic partition (0) or add a new topic partition (1) without an ID. FetchSessionHandler.Builder builder2 = handler.newBuilder(); -builder2.add(new TopicPartition("foo", partition), Uuid.ZERO_UUID, -new FetchRequest.PartitionData(10, 110, 210, Optional.empty())); +builder2.add(new TopicPartition("foo", partition), +new FetchRequest.PartitionData(Uuid.ZERO_UUID, 10, 110, 210, Optional.empty())); FetchSessionHandler.FetchRequestData data2 = builder2.build(); -// Should have the same session ID and next epoch, but can no longer use topic IDs. -// The receiving broker will close the session if we were previously using topic IDs. +// Should have the same session ID, and next epoch and can no longer use topic IDs. +// The receiving broker will handle closing the session. assertEquals(123, data2.metadata().sessionId(), "Did not use same session when " + testType); assertEquals(1, data2.metadata().epoch(), "Did not have correct epoch when " + testType); assertFalse(data2.canUseTopicIds()); }); } +@ParameterizedTest +@ValueSource(booleans = {true, false}) +public void testTopicIdReplaced(boolean fetchRequestUsesIds) { +TopicPartition tp = new TopicPartition("foo", 0); +FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); +FetchSessionHandler.Builder builder = handler.newBuilder(); +Uuid topicId1 = Uuid.randomUuid(); +builder.add(tp, +new FetchRequest.PartitionData(topicId1, 0, 100, 200, Optional.empty())); +FetchSessionHandler.FetchRequestData data = builder.build(); +assertMapsEqual(reqMap(new ReqEntry("foo", topicId1, 0, 0, 100, 200)), +data.toSend(), data.sessionPartitions()); +assertTrue(data.metadata().isFull()); +assertTrue(data.canUseTopicIds()); + +FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, +respMap(new RespEntry("foo", 0, Uuid.ZERO_UUID, 10, 20))); +handler.handleResponse(resp, (short) 12); Review comment: ah good catch. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743240526 ## File path: clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java ## @@ -428,82 +442,165 @@ public void testIdUsageRevokedOnIdDowngrade() { List partitions = Arrays.asList(0, 1); partitions.forEach(partition -> { String testType = partition == 0 ? "updating a partition" : "adding a new partition"; -Map topicIds = Collections.singletonMap("foo", Uuid.randomUuid()); +Uuid fooId = Uuid.randomUuid(); FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); FetchSessionHandler.Builder builder = handler.newBuilder(); -builder.add(new TopicPartition("foo", 0), topicIds.get("foo"), -new FetchRequest.PartitionData(0, 100, 200, Optional.empty())); +builder.add(new TopicPartition("foo", 0), +new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty())); FetchSessionHandler.FetchRequestData data = builder.build(); -assertMapsEqual(reqMap(new ReqEntry("foo", 0, 0, 100, 200)), +assertMapsEqual(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200)), data.toSend(), data.sessionPartitions()); assertTrue(data.metadata().isFull()); assertTrue(data.canUseTopicIds()); FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, -respMap(new RespEntry("foo", 0, 10, 20)), topicIds); +respMap(new RespEntry("foo", 0, fooId, 10, 20))); handler.handleResponse(resp, ApiKeys.FETCH.latestVersion()); // Try to remove a topic ID from an existing topic partition (0) or add a new topic partition (1) without an ID. FetchSessionHandler.Builder builder2 = handler.newBuilder(); -builder2.add(new TopicPartition("foo", partition), Uuid.ZERO_UUID, -new FetchRequest.PartitionData(10, 110, 210, Optional.empty())); +builder2.add(new TopicPartition("foo", partition), +new FetchRequest.PartitionData(Uuid.ZERO_UUID, 10, 110, 210, Optional.empty())); FetchSessionHandler.FetchRequestData data2 = builder2.build(); -// Should have the same session ID and next epoch, but can no longer use topic IDs. -// The receiving broker will close the session if we were previously using topic IDs. +// Should have the same session ID, and next epoch and can no longer use topic IDs. +// The receiving broker will handle closing the session. assertEquals(123, data2.metadata().sessionId(), "Did not use same session when " + testType); assertEquals(1, data2.metadata().epoch(), "Did not have correct epoch when " + testType); assertFalse(data2.canUseTopicIds()); }); } +@ParameterizedTest +@ValueSource(booleans = {true, false}) +public void testTopicIdReplaced(boolean fetchRequestUsesIds) { +TopicPartition tp = new TopicPartition("foo", 0); +FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); +FetchSessionHandler.Builder builder = handler.newBuilder(); +Uuid topicId1 = Uuid.randomUuid(); +builder.add(tp, +new FetchRequest.PartitionData(topicId1, 0, 100, 200, Optional.empty())); +FetchSessionHandler.FetchRequestData data = builder.build(); +assertMapsEqual(reqMap(new ReqEntry("foo", topicId1, 0, 0, 100, 200)), +data.toSend(), data.sessionPartitions()); +assertTrue(data.metadata().isFull()); +assertTrue(data.canUseTopicIds()); + +FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, +respMap(new RespEntry("foo", 0, Uuid.ZERO_UUID, 10, 20))); +handler.handleResponse(resp, (short) 12); + +// Try to add a new topic ID. +FetchSessionHandler.Builder builder2 = handler.newBuilder(); +Uuid topicId2 = Uuid.randomUuid(); +// Use the same data besides the topic ID. +FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId2, 0, 100, 200, Optional.empty()); +builder2.add(tp, partitionData); +FetchSessionHandler.FetchRequestData data2 = builder2.build(); +// The old topic ID partition should be in toReplace, and the new one should be in toSend. +assertMapsEqual(reqMap(new ReqEntry("foo", topicId2, 0, 0, 100, 200)), +data2.toSend(), data2.sessionPartitions()); +assertEquals(Collections.singletonList(new TopicIdPartition(topicId1, tp)), data2.toReplace()); Review comment: To clarify -- are you referring to a case where we upgraded? ie, it started with no ID in the first request and added one in the secon
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743241204 ## File path: clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java ## @@ -428,82 +442,165 @@ public void testIdUsageRevokedOnIdDowngrade() { List partitions = Arrays.asList(0, 1); partitions.forEach(partition -> { String testType = partition == 0 ? "updating a partition" : "adding a new partition"; -Map topicIds = Collections.singletonMap("foo", Uuid.randomUuid()); +Uuid fooId = Uuid.randomUuid(); FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); FetchSessionHandler.Builder builder = handler.newBuilder(); -builder.add(new TopicPartition("foo", 0), topicIds.get("foo"), -new FetchRequest.PartitionData(0, 100, 200, Optional.empty())); +builder.add(new TopicPartition("foo", 0), +new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty())); FetchSessionHandler.FetchRequestData data = builder.build(); -assertMapsEqual(reqMap(new ReqEntry("foo", 0, 0, 100, 200)), +assertMapsEqual(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200)), data.toSend(), data.sessionPartitions()); assertTrue(data.metadata().isFull()); assertTrue(data.canUseTopicIds()); FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, -respMap(new RespEntry("foo", 0, 10, 20)), topicIds); +respMap(new RespEntry("foo", 0, fooId, 10, 20))); handler.handleResponse(resp, ApiKeys.FETCH.latestVersion()); // Try to remove a topic ID from an existing topic partition (0) or add a new topic partition (1) without an ID. FetchSessionHandler.Builder builder2 = handler.newBuilder(); -builder2.add(new TopicPartition("foo", partition), Uuid.ZERO_UUID, -new FetchRequest.PartitionData(10, 110, 210, Optional.empty())); +builder2.add(new TopicPartition("foo", partition), +new FetchRequest.PartitionData(Uuid.ZERO_UUID, 10, 110, 210, Optional.empty())); FetchSessionHandler.FetchRequestData data2 = builder2.build(); -// Should have the same session ID and next epoch, but can no longer use topic IDs. -// The receiving broker will close the session if we were previously using topic IDs. +// Should have the same session ID, and next epoch and can no longer use topic IDs. +// The receiving broker will handle closing the session. assertEquals(123, data2.metadata().sessionId(), "Did not use same session when " + testType); assertEquals(1, data2.metadata().epoch(), "Did not have correct epoch when " + testType); assertFalse(data2.canUseTopicIds()); }); } +@ParameterizedTest +@ValueSource(booleans = {true, false}) +public void testTopicIdReplaced(boolean fetchRequestUsesIds) { +TopicPartition tp = new TopicPartition("foo", 0); +FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); +FetchSessionHandler.Builder builder = handler.newBuilder(); +Uuid topicId1 = Uuid.randomUuid(); +builder.add(tp, +new FetchRequest.PartitionData(topicId1, 0, 100, 200, Optional.empty())); +FetchSessionHandler.FetchRequestData data = builder.build(); +assertMapsEqual(reqMap(new ReqEntry("foo", topicId1, 0, 0, 100, 200)), +data.toSend(), data.sessionPartitions()); +assertTrue(data.metadata().isFull()); +assertTrue(data.canUseTopicIds()); + +FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, +respMap(new RespEntry("foo", 0, Uuid.ZERO_UUID, 10, 20))); +handler.handleResponse(resp, (short) 12); + +// Try to add a new topic ID. +FetchSessionHandler.Builder builder2 = handler.newBuilder(); +Uuid topicId2 = Uuid.randomUuid(); +// Use the same data besides the topic ID. +FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId2, 0, 100, 200, Optional.empty()); +builder2.add(tp, partitionData); +FetchSessionHandler.FetchRequestData data2 = builder2.build(); +// The old topic ID partition should be in toReplace, and the new one should be in toSend. +assertMapsEqual(reqMap(new ReqEntry("foo", topicId2, 0, 0, 100, 200)), +data2.toSend(), data2.sessionPartitions()); +assertEquals(Collections.singletonList(new TopicIdPartition(topicId1, tp)), data2.toReplace()); Review comment: I could theoretically check replace in the other test that checks multiple scenarios -- This is an automated message from the Apa
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743241513 ## File path: clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java ## @@ -428,82 +442,165 @@ public void testIdUsageRevokedOnIdDowngrade() { List partitions = Arrays.asList(0, 1); partitions.forEach(partition -> { String testType = partition == 0 ? "updating a partition" : "adding a new partition"; -Map topicIds = Collections.singletonMap("foo", Uuid.randomUuid()); +Uuid fooId = Uuid.randomUuid(); FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); FetchSessionHandler.Builder builder = handler.newBuilder(); -builder.add(new TopicPartition("foo", 0), topicIds.get("foo"), -new FetchRequest.PartitionData(0, 100, 200, Optional.empty())); +builder.add(new TopicPartition("foo", 0), +new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty())); FetchSessionHandler.FetchRequestData data = builder.build(); -assertMapsEqual(reqMap(new ReqEntry("foo", 0, 0, 100, 200)), +assertMapsEqual(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200)), data.toSend(), data.sessionPartitions()); assertTrue(data.metadata().isFull()); assertTrue(data.canUseTopicIds()); FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, -respMap(new RespEntry("foo", 0, 10, 20)), topicIds); +respMap(new RespEntry("foo", 0, fooId, 10, 20))); handler.handleResponse(resp, ApiKeys.FETCH.latestVersion()); // Try to remove a topic ID from an existing topic partition (0) or add a new topic partition (1) without an ID. FetchSessionHandler.Builder builder2 = handler.newBuilder(); -builder2.add(new TopicPartition("foo", partition), Uuid.ZERO_UUID, -new FetchRequest.PartitionData(10, 110, 210, Optional.empty())); +builder2.add(new TopicPartition("foo", partition), +new FetchRequest.PartitionData(Uuid.ZERO_UUID, 10, 110, 210, Optional.empty())); FetchSessionHandler.FetchRequestData data2 = builder2.build(); -// Should have the same session ID and next epoch, but can no longer use topic IDs. -// The receiving broker will close the session if we were previously using topic IDs. +// Should have the same session ID, and next epoch and can no longer use topic IDs. +// The receiving broker will handle closing the session. assertEquals(123, data2.metadata().sessionId(), "Did not use same session when " + testType); assertEquals(1, data2.metadata().epoch(), "Did not have correct epoch when " + testType); assertFalse(data2.canUseTopicIds()); }); } +@ParameterizedTest +@ValueSource(booleans = {true, false}) +public void testTopicIdReplaced(boolean fetchRequestUsesIds) { +TopicPartition tp = new TopicPartition("foo", 0); +FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); +FetchSessionHandler.Builder builder = handler.newBuilder(); +Uuid topicId1 = Uuid.randomUuid(); +builder.add(tp, +new FetchRequest.PartitionData(topicId1, 0, 100, 200, Optional.empty())); +FetchSessionHandler.FetchRequestData data = builder.build(); +assertMapsEqual(reqMap(new ReqEntry("foo", topicId1, 0, 0, 100, 200)), +data.toSend(), data.sessionPartitions()); +assertTrue(data.metadata().isFull()); +assertTrue(data.canUseTopicIds()); + +FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, +respMap(new RespEntry("foo", 0, Uuid.ZERO_UUID, 10, 20))); +handler.handleResponse(resp, (short) 12); + +// Try to add a new topic ID. +FetchSessionHandler.Builder builder2 = handler.newBuilder(); +Uuid topicId2 = Uuid.randomUuid(); +// Use the same data besides the topic ID. +FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId2, 0, 100, 200, Optional.empty()); +builder2.add(tp, partitionData); +FetchSessionHandler.FetchRequestData data2 = builder2.build(); +// The old topic ID partition should be in toReplace, and the new one should be in toSend. +assertMapsEqual(reqMap(new ReqEntry("foo", topicId2, 0, 0, 100, 200)), +data2.toSend(), data2.sessionPartitions()); +assertEquals(Collections.singletonList(new TopicIdPartition(topicId1, tp)), data2.toReplace()); +// Should have the same session ID, and next epoch and can use topic IDs. +assertEquals(123, data2.metadata().sessionId(), "Did not use same
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743242346 ## File path: clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java ## @@ -428,82 +442,165 @@ public void testIdUsageRevokedOnIdDowngrade() { List partitions = Arrays.asList(0, 1); partitions.forEach(partition -> { String testType = partition == 0 ? "updating a partition" : "adding a new partition"; -Map topicIds = Collections.singletonMap("foo", Uuid.randomUuid()); +Uuid fooId = Uuid.randomUuid(); FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); FetchSessionHandler.Builder builder = handler.newBuilder(); -builder.add(new TopicPartition("foo", 0), topicIds.get("foo"), -new FetchRequest.PartitionData(0, 100, 200, Optional.empty())); +builder.add(new TopicPartition("foo", 0), +new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty())); FetchSessionHandler.FetchRequestData data = builder.build(); -assertMapsEqual(reqMap(new ReqEntry("foo", 0, 0, 100, 200)), +assertMapsEqual(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200)), data.toSend(), data.sessionPartitions()); assertTrue(data.metadata().isFull()); assertTrue(data.canUseTopicIds()); FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, -respMap(new RespEntry("foo", 0, 10, 20)), topicIds); +respMap(new RespEntry("foo", 0, fooId, 10, 20))); handler.handleResponse(resp, ApiKeys.FETCH.latestVersion()); // Try to remove a topic ID from an existing topic partition (0) or add a new topic partition (1) without an ID. FetchSessionHandler.Builder builder2 = handler.newBuilder(); -builder2.add(new TopicPartition("foo", partition), Uuid.ZERO_UUID, -new FetchRequest.PartitionData(10, 110, 210, Optional.empty())); +builder2.add(new TopicPartition("foo", partition), +new FetchRequest.PartitionData(Uuid.ZERO_UUID, 10, 110, 210, Optional.empty())); FetchSessionHandler.FetchRequestData data2 = builder2.build(); -// Should have the same session ID and next epoch, but can no longer use topic IDs. -// The receiving broker will close the session if we were previously using topic IDs. +// Should have the same session ID, and next epoch and can no longer use topic IDs. +// The receiving broker will handle closing the session. assertEquals(123, data2.metadata().sessionId(), "Did not use same session when " + testType); assertEquals(1, data2.metadata().epoch(), "Did not have correct epoch when " + testType); assertFalse(data2.canUseTopicIds()); }); } +@ParameterizedTest +@ValueSource(booleans = {true, false}) +public void testTopicIdReplaced(boolean fetchRequestUsesIds) { +TopicPartition tp = new TopicPartition("foo", 0); +FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); +FetchSessionHandler.Builder builder = handler.newBuilder(); +Uuid topicId1 = Uuid.randomUuid(); +builder.add(tp, +new FetchRequest.PartitionData(topicId1, 0, 100, 200, Optional.empty())); +FetchSessionHandler.FetchRequestData data = builder.build(); +assertMapsEqual(reqMap(new ReqEntry("foo", topicId1, 0, 0, 100, 200)), +data.toSend(), data.sessionPartitions()); +assertTrue(data.metadata().isFull()); +assertTrue(data.canUseTopicIds()); + +FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, +respMap(new RespEntry("foo", 0, Uuid.ZERO_UUID, 10, 20))); +handler.handleResponse(resp, (short) 12); + +// Try to add a new topic ID. +FetchSessionHandler.Builder builder2 = handler.newBuilder(); +Uuid topicId2 = Uuid.randomUuid(); +// Use the same data besides the topic ID. +FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId2, 0, 100, 200, Optional.empty()); +builder2.add(tp, partitionData); +FetchSessionHandler.FetchRequestData data2 = builder2.build(); +// The old topic ID partition should be in toReplace, and the new one should be in toSend. +assertMapsEqual(reqMap(new ReqEntry("foo", topicId2, 0, 0, 100, 200)), +data2.toSend(), data2.sessionPartitions()); +assertEquals(Collections.singletonList(new TopicIdPartition(topicId1, tp)), data2.toReplace()); +// Should have the same session ID, and next epoch and can use topic IDs. +assertEquals(123, data2.metadata().sessionId(), "Did not use same
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743242849 ## File path: clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java ## @@ -2069,10 +2071,10 @@ public void testReturnRecordsDuringRebalance() throws InterruptedException { ConsumerMetadata metadata = createMetadata(subscription); MockClient client = new MockClient(time, metadata); ConsumerPartitionAssignor assignor = new CooperativeStickyAssignor(); -KafkaConsumer consumer = newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); - initMetadata(client, Utils.mkMap(Utils.mkEntry(topic, 1), Utils.mkEntry(topic2, 1), Utils.mkEntry(topic3, 1))); +KafkaConsumer consumer = newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); Review comment: It likely had something to do with how the mock client was handling metadata. But that may have been for the older version where we checked NodeApiVersion. I can try to switch it back. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743243111 ## File path: clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java ## @@ -2659,6 +2661,9 @@ private FetchResponse fetchResponse(TopicPartition partition, long fetchOffset, autoCommitIntervalMs, interceptors, throwOnStableOffsetNotSupported); +ApiVersions apiVersions = new ApiVersions(); +metadata.fetch().nodes().forEach(node -> +apiVersions.update(node.idString(), NodeApiVersions.create())); Review comment: Nope. Looks like another change I forgot to cleanup. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Commented] (KAFKA-13375) Kafka streams apps w/EOS unable to start at InitProducerId
[ https://issues.apache.org/jira/browse/KAFKA-13375?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17438957#comment-17438957 ] Lerh Chuan Low commented on KAFKA-13375: [~guozhang] Thank you for reading through the ticket! I hadn't seen that KIP yet and didn't realise this was a known issue, I'll digest. > Kafka streams apps w/EOS unable to start at InitProducerId > -- > > Key: KAFKA-13375 > URL: https://issues.apache.org/jira/browse/KAFKA-13375 > Project: Kafka > Issue Type: Bug > Components: core >Affects Versions: 2.8.0 >Reporter: Lerh Chuan Low >Priority: Major > > Hello, I'm wondering if this is a Kafka bug. Our environment setup is as > follows: > Kafka streams 2.8 - with *EXACTLY_ONCE* turned on (Not *EOS_BETA*, but I > don't think the changes introduced in EOS beta should affect this). > Transaction timeout = 60s. > Kafka broker 2.8. > We have this situation where we were doing a rolling restart of the broker to > apply some security changes. After we finished, 4 out of some 15 Stream Apps > are unable to start. They can never succeed, no matter what we do. > They fail with the error: > {code:java} > 2021-10-14 07:20:13,548 WARN > [srn-rec-feeder-802c18a1-9512-4a2a-8c2e-00e37550199d-StreamThread-3] > o.a.k.s.p.i.StreamsProducer stream-thread > [srn-rec-feeder-802c18a1-9512-4a2a-8c2e-00e37550199d-StreamThread-3] task > [0_6] Timeout exception caught trying to initialize transactions. The broker > is either slow or in bad state (like not having enough replicas) in > responding to the request, or the connection to broker was interrupted > sending the request or receiving the response. Will retry initializing the > task in the next loop. Consider overwriting max.block.ms to a larger value to > avoid timeout errors{code} > We found a previous Jira describing the issue here: > https://issues.apache.org/jira/browse/KAFKA-8803. It seems like back then > what people did was to rolling restart the brokers. We tried that - we > targeted the group coordinators for our failing apps, then transaction > coordinators, then all of them. It hasn't resolved our issue so far. > A few interesting things we've found so far: > - What I can see is that all the failing apps only fail on certain > partitions. E.g for the app above, only partition 6 never succeeds. Partition > 6 shares the same coordinator as some of the other partitions and those work, > so it seems like the issue isn't related to broker memory state. > - All the failing apps have a message similar to this > {code:java} > [2021-10-14 00:54:51,569] INFO [Transaction Marker Request Completion Handler > 103]: Sending srn-rec-feeder-0_6's transaction marker for partition > srn-bot-003-14 has permanently failed with error > org.apache.kafka.common.errors.InvalidProducerEpochException with the current > coordinator epoch 143; cancel sending any more transaction markers > TxnMarkerEntry{producerId=7001, producerEpoch=610, coordinatorEpoch=143, > result=ABORT, partitions=[srn-bot-003-14]} to the brokers > (kafka.coordinator.transaction.TransactionMarkerRequestCompletionHandler) > {code} > While we were restarting the brokers. They all failed shortly after. No other > consumer groups for the other working partitions/working stream apps logged > this message. > On digging around in git blame and reading through the source, it looks like > this is meant to be benign. > - We tried DEBUG logging for the TransactionCoordinator and > TransactionStateManager. We can see (assigner is a functioning app) > {code:java} > [2021-10-14 06:48:23,813] DEBUG [TransactionCoordinator id=105] Returning > CONCURRENT_TRANSACTIONS error code to client for srn-assigner-0_14's > AddPartitions request (kafka.coordinator.transaction.TransactionCoordinator) > {code} > I've seen those before during steady state. I do believe they are benign. We > never see it for the problematic partitions/consumer groups for some reason. > - We tried turning on TRACE for KafkaApis. We can see > {code:java} > [2021-10-14 06:56:58,408] TRACE [KafkaApi-105] Completed srn-rec-feeder-0_6's > InitProducerIdRequest with result > InitProducerIdResult(-1,-1,CONCURRENT_TRANSACTIONS) from client > srn-rec-feeder-802c18a1-9512-4a2a-8c2e-00e37550199d-StreamThread-4-0_6-producer. > (kafka.server.KafkaApis) {code} > It starts to make me wonder if there's a situation where Kafka is unable to > abort the transactions if there is never any success in initializing a > producer ID. But this is diving deep into insider knowledge territory that I > simply don't have. I'm wondering if anyone with more knowledge of how > transactions work can shed some light here if we are in the wrong path, or if > there's any way to restore operations at
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743243975 ## File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala ## @@ -374,7 +374,7 @@ abstract class AbstractFetcherThread(name: String, } } } catch { -case ime@( _: CorruptRecordException | _: InvalidRecordException) => +case ime@(_: CorruptRecordException | _: InvalidRecordException) => Review comment: > append when the controller fails over to an older IBP during an upgrade. I think I'm misunderstanding something here. Did you mean to say append? -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743243975 ## File path: core/src/main/scala/kafka/server/AbstractFetcherThread.scala ## @@ -374,7 +374,7 @@ abstract class AbstractFetcherThread(name: String, } } } catch { -case ime@( _: CorruptRecordException | _: InvalidRecordException) => +case ime@(_: CorruptRecordException | _: InvalidRecordException) => Review comment: > append when the controller fails over to an older IBP during an upgrade. I think I'm misunderstanding something here. Did you mean to say append? -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] RivenSun2 commented on pull request #11461: KAFKA-13422: Add verification of duplicate configuration for each type of LoginModule in JaasConfigFile
RivenSun2 commented on pull request #11461: URL: https://github.com/apache/kafka/pull/11461#issuecomment-961575399 @guozhangwang @ijuma please help check this PR when available. Thank you. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743349235 ## File path: clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java ## @@ -4814,6 +4842,7 @@ private void buildDependencies(MetricConfig metricConfig, metadata = new ConsumerMetadata(0, metadataExpireMs, false, false, subscriptions, logContext, new ClusterResourceListeners()); client = new MockClient(time, metadata); +client.setNodeApiVersions(NodeApiVersions.create()); metrics = new Metrics(metricConfig, time); consumerClient = new ConsumerNetworkClient(logContext, client, metadata, time, 100, 1000, Integer.MAX_VALUE); Review comment: Are you referring to how we changed UNKNOWN_TOPIC_ID and INCONSISTENT_TOPIC_ID? For these cases we have testFetchInconsistentTopicId and testFetchUnknownTopicId which check that we update the metadata for a partition level error. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743349370 ## File path: clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java ## @@ -4814,6 +4842,7 @@ private void buildDependencies(MetricConfig metricConfig, metadata = new ConsumerMetadata(0, metadataExpireMs, false, false, subscriptions, logContext, new ClusterResourceListeners()); client = new MockClient(time, metadata); +client.setNodeApiVersions(NodeApiVersions.create()); metrics = new Metrics(metricConfig, time); consumerClient = new ConsumerNetworkClient(logContext, client, metadata, time, 100, 1000, Integer.MAX_VALUE); Review comment: These tests changed from returning a top level error to partition level error. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743350813 ## File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala ## @@ -3530,37 +3534,37 @@ class KafkaApisTest { def testSizeOfThrottledPartitions(): Unit = { val topicNames = new util.HashMap[Uuid, String] val topicIds = new util.HashMap[String, Uuid]() -def fetchResponse(data: Map[TopicPartition, String]): FetchResponse = { - val responseData = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData]( +def fetchResponse(data: Map[TopicIdPartition, String]): FetchResponse = { + val responseData = new util.LinkedHashMap[TopicIdPartition, FetchResponseData.PartitionData]( data.map { case (tp, raw) => tp -> new FetchResponseData.PartitionData() -.setPartitionIndex(tp.partition) +.setPartitionIndex(tp.topicPartition.partition) .setHighWatermark(105) .setLastStableOffset(105) .setLogStartOffset(0) .setRecords(MemoryRecords.withRecords(CompressionType.NONE, new SimpleRecord(100, raw.getBytes(StandardCharsets.UTF_8 }.toMap.asJava) data.foreach{case (tp, _) => -val id = Uuid.randomUuid() -topicIds.put(tp.topic(), id) -topicNames.put(id, tp.topic()) +topicIds.put(tp.topicPartition.topic, tp.topicId) +topicNames.put(tp.topicId, tp.topicPartition.topic) } - FetchResponse.of(Errors.NONE, 100, 100, responseData, topicIds) + FetchResponse.of(Errors.NONE, 100, 100, responseData) } -val throttledPartition = new TopicPartition("throttledData", 0) +val throttledPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("throttledData", 0)) val throttledData = Map(throttledPartition -> "throttledData") val expectedSize = FetchResponse.sizeOf(FetchResponseData.HIGHEST_SUPPORTED_VERSION, - fetchResponse(throttledData).responseData(topicNames, FetchResponseData.HIGHEST_SUPPORTED_VERSION).entrySet.iterator, topicIds) + fetchResponse(throttledData).responseData(topicNames, FetchResponseData.HIGHEST_SUPPORTED_VERSION).entrySet.asScala.map( entry => + (new TopicIdPartition(Uuid.ZERO_UUID, entry.getKey), entry.getValue)).toMap.asJava.entrySet.iterator) -val response = fetchResponse(throttledData ++ Map(new TopicPartition("nonThrottledData", 0) -> "nonThrottledData")) +val response = fetchResponse(throttledData ++ Map(new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("nonThrottledData", 0)) -> "nonThrottledData")) val quota = Mockito.mock(classOf[ReplicationQuotaManager]) Mockito.when(quota.isThrottled(ArgumentMatchers.any(classOf[TopicPartition]))) - .thenAnswer(invocation => throttledPartition == invocation.getArgument(0).asInstanceOf[TopicPartition]) + .thenAnswer(invocation => throttledPartition.topicPartition == invocation.getArgument(0).asInstanceOf[TopicPartition]) -assertEquals(expectedSize, KafkaApis.sizeOfThrottledPartitions(FetchResponseData.HIGHEST_SUPPORTED_VERSION, response, quota, topicIds)) +assertEquals(expectedSize, KafkaApis.sizeOfThrottledPartitions(FetchResponseData.HIGHEST_SUPPORTED_VERSION, response, quota)) } @Test Review comment: What logic are we thinking? Checking that the unresolved topics are handled correctly? -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743351017 ## File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala ## @@ -1361,102 +1542,113 @@ class FetchSessionTest { val resp4 = context2.updateAndGenerateResponseData(respData) assertEquals(Errors.NONE, resp4.error) assertEquals(resp1.sessionId, resp4.sessionId) -assertEquals(Utils.mkSet(tp1, tp2), resp4.responseData(topicNames, request2.version).keySet) +assertEquals(Utils.mkSet(tp1.topicPartition, tp2.topicPartition), resp4.responseData(topicNames, request2.version).keySet) } @Test def testDeprioritizesPartitionsWithRecordsOnly(): Unit = { val time = new MockTime() val cache = new FetchSessionCache(10, 1000) val fetchManager = new FetchManager(time, cache) -val tp1 = new TopicPartition("foo", 1) -val tp2 = new TopicPartition("bar", 2) -val tp3 = new TopicPartition("zar", 3) val topicIds = Map("foo" -> Uuid.randomUuid(), "bar" -> Uuid.randomUuid(), "zar" -> Uuid.randomUuid()).asJava val topicNames = topicIds.asScala.map(_.swap).asJava +val tp1 = new TopicIdPartition(topicIds.get("foo"), new TopicPartition("foo", 1)) +val tp2 = new TopicIdPartition(topicIds.get("bar"), new TopicPartition("bar", 2)) +val tp3 = new TopicIdPartition(topicIds.get("zar"), new TopicPartition("zar", 3)) -val reqData = new util.LinkedHashMap[TopicPartition, FetchRequest.PartitionData] -reqData.put(tp1, new FetchRequest.PartitionData(100, 0, 1000, Optional.of(5), Optional.of(4))) -reqData.put(tp2, new FetchRequest.PartitionData(100, 0, 1000, Optional.of(5), Optional.of(4))) -reqData.put(tp3, new FetchRequest.PartitionData(100, 0, 1000, Optional.of(5), Optional.of(4))) +val reqData = new util.LinkedHashMap[TopicIdPartition, FetchRequest.PartitionData] +reqData.put(tp1, new FetchRequest.PartitionData(topicIds.get("foo"), 100, 0, 1000, Optional.of(5), Optional.of(4))) +reqData.put(tp2, new FetchRequest.PartitionData(topicIds.get("bar"), 100, 0, 1000, Optional.of(5), Optional.of(4))) +reqData.put(tp3, new FetchRequest.PartitionData(topicIds.get("zar"), 100, 0, 1000, Optional.of(5), Optional.of(4))) // Full fetch context returns all partitions in the response val context1 = fetchManager.newContext(ApiKeys.FETCH.latestVersion(), JFetchMetadata.INITIAL, false, - reqData, Collections.emptyList(), topicIds) + reqData, Collections.emptyList(), topicNames) assertEquals(classOf[FullFetchContext], context1.getClass) -val respData1 = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData] +val respData1 = new util.LinkedHashMap[TopicIdPartition, FetchResponseData.PartitionData] respData1.put(tp1, new FetchResponseData.PartitionData() - .setPartitionIndex(tp1.partition) + .setPartitionIndex(tp1.topicPartition.partition) .setHighWatermark(50) .setLastStableOffset(50) .setLogStartOffset(0)) respData1.put(tp2, new FetchResponseData.PartitionData() - .setPartitionIndex(tp2.partition) + .setPartitionIndex(tp2.topicPartition.partition) .setHighWatermark(50) .setLastStableOffset(50) .setLogStartOffset(0)) respData1.put(tp3, new FetchResponseData.PartitionData() - .setPartitionIndex(tp3.partition) + .setPartitionIndex(tp3.topicPartition.partition) .setHighWatermark(50) .setLastStableOffset(50) .setLogStartOffset(0)) val resp1 = context1.updateAndGenerateResponseData(respData1) assertEquals(Errors.NONE, resp1.error) assertNotEquals(INVALID_SESSION_ID, resp1.sessionId) -assertEquals(Utils.mkSet(tp1, tp2, tp3), resp1.responseData(topicNames, ApiKeys.FETCH.latestVersion()).keySet()) +assertEquals(Utils.mkSet(tp1.topicPartition, tp2.topicPartition, tp3.topicPartition), resp1.responseData(topicNames, ApiKeys.FETCH.latestVersion()).keySet()) // Incremental fetch context returns partitions with changes but only deprioritizes // the partitions with records val context2 = fetchManager.newContext(ApiKeys.FETCH.latestVersion(), new JFetchMetadata(resp1.sessionId, 1), false, - reqData, Collections.emptyList(), topicIds) + reqData, Collections.emptyList(), topicNames) assertEquals(classOf[IncrementalFetchContext], context2.getClass) // Partitions are ordered in the session as per last response assertPartitionsOrder(context2, Seq(tp1, tp2, tp3)) // Response is empty -val respData2 = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData] +val respData2 = new util.LinkedHashMap[TopicIdPartition, FetchResponseData.PartitionData] val resp2 = context2.updateAndGenerateResponseData(respData2) assertEquals(Errors.NONE, resp2.error) assertEquals(resp1.sessionId, resp2.sessionId) assertEquals(Collections.empty
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743354926 ## File path: core/src/main/scala/kafka/server/FetchSession.scala ## @@ -163,18 +173,35 @@ class CachedPartition(val topic: String, mustRespond } - override def hashCode: Int = Objects.hash(new TopicPartition(topic, partition), topicId) + /** + * We have different equality checks depending on whether topic IDs are used. + * This means we need a different hash function as well. We use name to calculate the hash if the ID is zero and unused. + * Otherwise, we use the topic ID in the hash calculation. + * + * @return the hash code for the CachedPartition depending on what request version we are using. + */ + override def hashCode: Int = if (topicId != Uuid.ZERO_UUID) (31 * partition) + topicId.hashCode else +(31 * partition) + topic.hashCode def canEqual(that: Any): Boolean = that.isInstanceOf[CachedPartition] + /** + * We have different equality checks depending on whether topic IDs are used. + * + * This is because when we use topic IDs, a partition with a given ID and an unknown name is the same as a partition with that + * ID and a known name. This means we can only use topic ID and partition when determining equality. + * + * On the other hand, if we are using topic names, all IDs are zero. This means we can only use topic name and partition + * when determining equality. + */ override def equals(that: Any): Boolean = that match { case that: CachedPartition => this.eq(that) || (that.canEqual(this) && Review comment: Hmm. I'm not quite sure why this would not make sense. I believe it is checking the types are correct. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743355748 ## File path: core/src/main/scala/kafka/server/KafkaApis.scala ## @@ -708,40 +701,41 @@ class KafkaApis(val requestChannel: RequestChannel, None } -val erroneous = mutable.ArrayBuffer[(TopicPartition, FetchResponseData.PartitionData)]() -val interesting = mutable.ArrayBuffer[(TopicPartition, FetchRequest.PartitionData)]() -val sessionTopicIds = mutable.Map[String, Uuid]() +val erroneous = mutable.ArrayBuffer[(TopicIdPartition, FetchResponseData.PartitionData)]() +val interesting = mutable.ArrayBuffer[(TopicIdPartition, FetchRequest.PartitionData)]() if (fetchRequest.isFromFollower) { // The follower must have ClusterAction on ClusterResource in order to fetch partition data. if (authHelper.authorize(request.context, CLUSTER_ACTION, CLUSTER, CLUSTER_NAME)) { -fetchContext.foreachPartition { (topicPartition, topicId, data) => - sessionTopicIds.put(topicPartition.topic(), topicId) - if (!metadataCache.contains(topicPartition)) -erroneous += topicPartition -> FetchResponse.partitionResponse(topicPartition.partition, Errors.UNKNOWN_TOPIC_OR_PARTITION) +fetchContext.foreachPartition { (topicIdPartition, data) => + if (topicIdPartition.topicPartition.topic == null ) +erroneous += topicIdPartition -> FetchResponse.partitionResponse(topicIdPartition, Errors.UNKNOWN_TOPIC_ID) + else if (!metadataCache.contains(topicIdPartition.topicPartition)) +erroneous += topicIdPartition -> FetchResponse.partitionResponse(topicIdPartition, Errors.UNKNOWN_TOPIC_OR_PARTITION) else -interesting += (topicPartition -> data) +interesting += (topicIdPartition -> data) } } else { -fetchContext.foreachPartition { (part, topicId, _) => - sessionTopicIds.put(part.topic(), topicId) - erroneous += part -> FetchResponse.partitionResponse(part.partition, Errors.TOPIC_AUTHORIZATION_FAILED) +fetchContext.foreachPartition { (topicIdPartition, _) => + erroneous += topicIdPartition -> FetchResponse.partitionResponse(topicIdPartition, Errors.TOPIC_AUTHORIZATION_FAILED) Review comment: I think we would want to keep the authorization error. Since it just logs a message. The UNKNOWN_TOPIC_ID error would request a metadata update which doesn't make sense when there is an authorization error. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743357376 ## File path: clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java ## @@ -428,82 +442,165 @@ public void testIdUsageRevokedOnIdDowngrade() { List partitions = Arrays.asList(0, 1); partitions.forEach(partition -> { String testType = partition == 0 ? "updating a partition" : "adding a new partition"; -Map topicIds = Collections.singletonMap("foo", Uuid.randomUuid()); +Uuid fooId = Uuid.randomUuid(); FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); FetchSessionHandler.Builder builder = handler.newBuilder(); -builder.add(new TopicPartition("foo", 0), topicIds.get("foo"), -new FetchRequest.PartitionData(0, 100, 200, Optional.empty())); +builder.add(new TopicPartition("foo", 0), +new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty())); FetchSessionHandler.FetchRequestData data = builder.build(); -assertMapsEqual(reqMap(new ReqEntry("foo", 0, 0, 100, 200)), +assertMapsEqual(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200)), data.toSend(), data.sessionPartitions()); assertTrue(data.metadata().isFull()); assertTrue(data.canUseTopicIds()); FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, -respMap(new RespEntry("foo", 0, 10, 20)), topicIds); +respMap(new RespEntry("foo", 0, fooId, 10, 20))); handler.handleResponse(resp, ApiKeys.FETCH.latestVersion()); // Try to remove a topic ID from an existing topic partition (0) or add a new topic partition (1) without an ID. FetchSessionHandler.Builder builder2 = handler.newBuilder(); -builder2.add(new TopicPartition("foo", partition), Uuid.ZERO_UUID, -new FetchRequest.PartitionData(10, 110, 210, Optional.empty())); +builder2.add(new TopicPartition("foo", partition), +new FetchRequest.PartitionData(Uuid.ZERO_UUID, 10, 110, 210, Optional.empty())); FetchSessionHandler.FetchRequestData data2 = builder2.build(); -// Should have the same session ID and next epoch, but can no longer use topic IDs. -// The receiving broker will close the session if we were previously using topic IDs. +// Should have the same session ID, and next epoch and can no longer use topic IDs. +// The receiving broker will handle closing the session. assertEquals(123, data2.metadata().sessionId(), "Did not use same session when " + testType); assertEquals(1, data2.metadata().epoch(), "Did not have correct epoch when " + testType); assertFalse(data2.canUseTopicIds()); }); } +@ParameterizedTest +@ValueSource(booleans = {true, false}) +public void testTopicIdReplaced(boolean fetchRequestUsesIds) { +TopicPartition tp = new TopicPartition("foo", 0); +FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); +FetchSessionHandler.Builder builder = handler.newBuilder(); +Uuid topicId1 = Uuid.randomUuid(); +builder.add(tp, +new FetchRequest.PartitionData(topicId1, 0, 100, 200, Optional.empty())); +FetchSessionHandler.FetchRequestData data = builder.build(); +assertMapsEqual(reqMap(new ReqEntry("foo", topicId1, 0, 0, 100, 200)), +data.toSend(), data.sessionPartitions()); +assertTrue(data.metadata().isFull()); +assertTrue(data.canUseTopicIds()); + +FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, +respMap(new RespEntry("foo", 0, Uuid.ZERO_UUID, 10, 20))); +handler.handleResponse(resp, (short) 12); + +// Try to add a new topic ID. +FetchSessionHandler.Builder builder2 = handler.newBuilder(); +Uuid topicId2 = Uuid.randomUuid(); +// Use the same data besides the topic ID. +FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId2, 0, 100, 200, Optional.empty()); +builder2.add(tp, partitionData); +FetchSessionHandler.FetchRequestData data2 = builder2.build(); +// The old topic ID partition should be in toReplace, and the new one should be in toSend. +assertMapsEqual(reqMap(new ReqEntry("foo", topicId2, 0, 0, 100, 200)), +data2.toSend(), data2.sessionPartitions()); +assertEquals(Collections.singletonList(new TopicIdPartition(topicId1, tp)), data2.toReplace()); Review comment: Or did you mean just adding something like `assertEquals(fetchRequestUsesIds, data2.toReplace().size() > 0);` -- This is an au
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743357376 ## File path: clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java ## @@ -428,82 +442,165 @@ public void testIdUsageRevokedOnIdDowngrade() { List partitions = Arrays.asList(0, 1); partitions.forEach(partition -> { String testType = partition == 0 ? "updating a partition" : "adding a new partition"; -Map topicIds = Collections.singletonMap("foo", Uuid.randomUuid()); +Uuid fooId = Uuid.randomUuid(); FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); FetchSessionHandler.Builder builder = handler.newBuilder(); -builder.add(new TopicPartition("foo", 0), topicIds.get("foo"), -new FetchRequest.PartitionData(0, 100, 200, Optional.empty())); +builder.add(new TopicPartition("foo", 0), +new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty())); FetchSessionHandler.FetchRequestData data = builder.build(); -assertMapsEqual(reqMap(new ReqEntry("foo", 0, 0, 100, 200)), +assertMapsEqual(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200)), data.toSend(), data.sessionPartitions()); assertTrue(data.metadata().isFull()); assertTrue(data.canUseTopicIds()); FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, -respMap(new RespEntry("foo", 0, 10, 20)), topicIds); +respMap(new RespEntry("foo", 0, fooId, 10, 20))); handler.handleResponse(resp, ApiKeys.FETCH.latestVersion()); // Try to remove a topic ID from an existing topic partition (0) or add a new topic partition (1) without an ID. FetchSessionHandler.Builder builder2 = handler.newBuilder(); -builder2.add(new TopicPartition("foo", partition), Uuid.ZERO_UUID, -new FetchRequest.PartitionData(10, 110, 210, Optional.empty())); +builder2.add(new TopicPartition("foo", partition), +new FetchRequest.PartitionData(Uuid.ZERO_UUID, 10, 110, 210, Optional.empty())); FetchSessionHandler.FetchRequestData data2 = builder2.build(); -// Should have the same session ID and next epoch, but can no longer use topic IDs. -// The receiving broker will close the session if we were previously using topic IDs. +// Should have the same session ID, and next epoch and can no longer use topic IDs. +// The receiving broker will handle closing the session. assertEquals(123, data2.metadata().sessionId(), "Did not use same session when " + testType); assertEquals(1, data2.metadata().epoch(), "Did not have correct epoch when " + testType); assertFalse(data2.canUseTopicIds()); }); } +@ParameterizedTest +@ValueSource(booleans = {true, false}) +public void testTopicIdReplaced(boolean fetchRequestUsesIds) { +TopicPartition tp = new TopicPartition("foo", 0); +FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); +FetchSessionHandler.Builder builder = handler.newBuilder(); +Uuid topicId1 = Uuid.randomUuid(); +builder.add(tp, +new FetchRequest.PartitionData(topicId1, 0, 100, 200, Optional.empty())); +FetchSessionHandler.FetchRequestData data = builder.build(); +assertMapsEqual(reqMap(new ReqEntry("foo", topicId1, 0, 0, 100, 200)), +data.toSend(), data.sessionPartitions()); +assertTrue(data.metadata().isFull()); +assertTrue(data.canUseTopicIds()); + +FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, +respMap(new RespEntry("foo", 0, Uuid.ZERO_UUID, 10, 20))); +handler.handleResponse(resp, (short) 12); + +// Try to add a new topic ID. +FetchSessionHandler.Builder builder2 = handler.newBuilder(); +Uuid topicId2 = Uuid.randomUuid(); +// Use the same data besides the topic ID. +FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId2, 0, 100, 200, Optional.empty()); +builder2.add(tp, partitionData); +FetchSessionHandler.FetchRequestData data2 = builder2.build(); +// The old topic ID partition should be in toReplace, and the new one should be in toSend. +assertMapsEqual(reqMap(new ReqEntry("foo", topicId2, 0, 0, 100, 200)), +data2.toSend(), data2.sessionPartitions()); +assertEquals(Collections.singletonList(new TopicIdPartition(topicId1, tp)), data2.toReplace()); Review comment: Or did you mean just adding something like `assertEquals(fetchRequestUsesIds, data2.toReplace().size() > 0);` -- This is an au
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743360648 ## File path: clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java ## @@ -428,82 +442,165 @@ public void testIdUsageRevokedOnIdDowngrade() { List partitions = Arrays.asList(0, 1); partitions.forEach(partition -> { String testType = partition == 0 ? "updating a partition" : "adding a new partition"; -Map topicIds = Collections.singletonMap("foo", Uuid.randomUuid()); +Uuid fooId = Uuid.randomUuid(); FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); FetchSessionHandler.Builder builder = handler.newBuilder(); -builder.add(new TopicPartition("foo", 0), topicIds.get("foo"), -new FetchRequest.PartitionData(0, 100, 200, Optional.empty())); +builder.add(new TopicPartition("foo", 0), +new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty())); FetchSessionHandler.FetchRequestData data = builder.build(); -assertMapsEqual(reqMap(new ReqEntry("foo", 0, 0, 100, 200)), +assertMapsEqual(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200)), data.toSend(), data.sessionPartitions()); assertTrue(data.metadata().isFull()); assertTrue(data.canUseTopicIds()); FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, -respMap(new RespEntry("foo", 0, 10, 20)), topicIds); +respMap(new RespEntry("foo", 0, fooId, 10, 20))); handler.handleResponse(resp, ApiKeys.FETCH.latestVersion()); // Try to remove a topic ID from an existing topic partition (0) or add a new topic partition (1) without an ID. FetchSessionHandler.Builder builder2 = handler.newBuilder(); -builder2.add(new TopicPartition("foo", partition), Uuid.ZERO_UUID, -new FetchRequest.PartitionData(10, 110, 210, Optional.empty())); +builder2.add(new TopicPartition("foo", partition), +new FetchRequest.PartitionData(Uuid.ZERO_UUID, 10, 110, 210, Optional.empty())); FetchSessionHandler.FetchRequestData data2 = builder2.build(); -// Should have the same session ID and next epoch, but can no longer use topic IDs. -// The receiving broker will close the session if we were previously using topic IDs. +// Should have the same session ID, and next epoch and can no longer use topic IDs. +// The receiving broker will handle closing the session. assertEquals(123, data2.metadata().sessionId(), "Did not use same session when " + testType); assertEquals(1, data2.metadata().epoch(), "Did not have correct epoch when " + testType); assertFalse(data2.canUseTopicIds()); }); } +@ParameterizedTest +@ValueSource(booleans = {true, false}) +public void testTopicIdReplaced(boolean fetchRequestUsesIds) { +TopicPartition tp = new TopicPartition("foo", 0); +FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1); +FetchSessionHandler.Builder builder = handler.newBuilder(); +Uuid topicId1 = Uuid.randomUuid(); +builder.add(tp, +new FetchRequest.PartitionData(topicId1, 0, 100, 200, Optional.empty())); +FetchSessionHandler.FetchRequestData data = builder.build(); +assertMapsEqual(reqMap(new ReqEntry("foo", topicId1, 0, 0, 100, 200)), +data.toSend(), data.sessionPartitions()); +assertTrue(data.metadata().isFull()); +assertTrue(data.canUseTopicIds()); + +FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, +respMap(new RespEntry("foo", 0, Uuid.ZERO_UUID, 10, 20))); +handler.handleResponse(resp, (short) 12); + +// Try to add a new topic ID. +FetchSessionHandler.Builder builder2 = handler.newBuilder(); +Uuid topicId2 = Uuid.randomUuid(); +// Use the same data besides the topic ID. +FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId2, 0, 100, 200, Optional.empty()); +builder2.add(tp, partitionData); +FetchSessionHandler.FetchRequestData data2 = builder2.build(); +// The old topic ID partition should be in toReplace, and the new one should be in toSend. +assertMapsEqual(reqMap(new ReqEntry("foo", topicId2, 0, 0, 100, 200)), +data2.toSend(), data2.sessionPartitions()); +assertEquals(Collections.singletonList(new TopicIdPartition(topicId1, tp)), data2.toReplace()); Review comment: Or are you just referring to a case where we don't ever have topic IDs? -- This is an automated message from the Apache Git Servic
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743361897 ## File path: clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java ## @@ -2069,10 +2071,10 @@ public void testReturnRecordsDuringRebalance() throws InterruptedException { ConsumerMetadata metadata = createMetadata(subscription); MockClient client = new MockClient(time, metadata); ConsumerPartitionAssignor assignor = new CooperativeStickyAssignor(); -KafkaConsumer consumer = newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); - initMetadata(client, Utils.mkMap(Utils.mkEntry(topic, 1), Utils.mkEntry(topic2, 1), Utils.mkEntry(topic3, 1))); +KafkaConsumer consumer = newConsumer(time, client, subscription, metadata, assignor, true, groupInstanceId); Review comment: Looks like most of these changes were done by this commit: https://github.com/apache/kafka/pull/11331/commits/32c6297adb685f1863b8c7eb85f2f0965853a9f8 so I can remove them pretty easily. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743365481 ## File path: core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala ## @@ -1091,18 +1089,20 @@ class AbstractFetcherThreadTest { override def buildFetch(partitionMap: Map[TopicPartition, PartitionFetchState]): ResultWithPartitions[Option[ReplicaFetch]] = { val fetchData = mutable.Map.empty[TopicPartition, FetchRequest.PartitionData] - partitionMap.foreach { case (partition, state) => + partitionMap.foreach { case (partition, state) => 0 +.equals(0) Review comment: I have no idea why this is here. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743369512 ## File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala ## @@ -659,88 +670,108 @@ class FetchSessionTest { } @Test - def testIncrementalFetchSessionWithIdsWhenSessionDoesNotUseIds() : Unit = { + def testFetchSessionWithUnknownId(): Unit = { val time = new MockTime() val cache = new FetchSessionCache(10, 1000) val fetchManager = new FetchManager(time, cache) -val topicIds = new util.HashMap[String, Uuid]() -val topicNames = new util.HashMap[Uuid, String]() +val topicNames = Map(Uuid.randomUuid() -> "foo", Uuid.randomUuid() -> "bar").asJava +val topicIds = topicNames.asScala.map(_.swap).asJava +val foo0 = new TopicIdPartition(topicIds.getOrDefault("foo", Uuid.ZERO_UUID), new TopicPartition("foo", 0)) +val foo1 = new TopicIdPartition(topicIds.getOrDefault("foo", Uuid.ZERO_UUID), new TopicPartition("foo", 1)) +val emptyFoo0 = new TopicIdPartition(topicIds.getOrDefault("foo", Uuid.ZERO_UUID), new TopicPartition(null, 0)) +val emptyFoo1 = new TopicIdPartition(topicIds.getOrDefault("foo", Uuid.ZERO_UUID), new TopicPartition(null, 1)) -// Create a new fetch session with foo-0 +// Create a new fetch session with foo-0 and foo-1 val reqData1 = new util.LinkedHashMap[TopicPartition, FetchRequest.PartitionData] -reqData1.put(new TopicPartition("foo", 0), new FetchRequest.PartitionData(0, 0, 100, +reqData1.put(foo0.topicPartition, new FetchRequest.PartitionData(foo0.topicId, 0, 0, 100, Optional.empty())) -val request1 = createRequestWithoutTopicIds(JFetchMetadata.INITIAL, reqData1, topicIds, EMPTY_PART_LIST, false) -// Start a fetch session using a request version that does not use topic IDs. +reqData1.put(foo1.topicPartition, new FetchRequest.PartitionData(foo1.topicId,10, 0, 100, + Optional.empty())) +val request1 = createRequest(JFetchMetadata.INITIAL, reqData1, EMPTY_PART_LIST, false) +// Simulate unknown topic ID for foo. +val topicNamesOnlyBar = Collections.singletonMap(topicIds.get("bar"), "bar") +// We should not throw error since we have an older request version. val context1 = fetchManager.newContext( request1.version, request1.metadata, request1.isFromFollower, - request1.fetchData(topicNames), - request1.forgottenTopics(topicNames), - topicIds + request1.fetchData(topicNamesOnlyBar), + request1.forgottenTopics(topicNamesOnlyBar), + topicNamesOnlyBar ) assertEquals(classOf[FullFetchContext], context1.getClass) -val respData1 = new util.LinkedHashMap[TopicPartition, FetchResponseData.PartitionData] -respData1.put(new TopicPartition("foo", 0), new FetchResponseData.PartitionData() +val respData1 = new util.LinkedHashMap[TopicIdPartition, FetchResponseData.PartitionData] Review comment: Not quite sure what you meant here but I added this for now: `context1.foreachPartition((topicIdPartition, _) => assertEquals(topicIds.get("foo"), topicIdPartition.topicId))` -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743370419 ## File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala ## @@ -659,88 +670,108 @@ class FetchSessionTest { } @Test - def testIncrementalFetchSessionWithIdsWhenSessionDoesNotUseIds() : Unit = { + def testFetchSessionWithUnknownId(): Unit = { val time = new MockTime() val cache = new FetchSessionCache(10, 1000) val fetchManager = new FetchManager(time, cache) -val topicIds = new util.HashMap[String, Uuid]() -val topicNames = new util.HashMap[Uuid, String]() +val topicNames = Map(Uuid.randomUuid() -> "foo", Uuid.randomUuid() -> "bar").asJava Review comment: We could do that, but then this check will be a bit more complicated. `context2.foreachPartition((topicIdPartition, _) => assertEquals(topicNames.get(topicIdPartition.topicId), topicIdPartition.topic))` -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743370708 ## File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala ## @@ -794,22 +793,23 @@ class ReplicaManagerTest { // We receive one valid request from the follower and replica state is updated var successfulFetch: Option[FetchPartitionData] = None - def callback(response: Seq[(TopicPartition, FetchPartitionData)]): Unit = { -successfulFetch = response.headOption.filter { case (topicPartition, _) => topicPartition == tp }.map { case (_, data) => data } + def callback(response: Seq[(TopicIdPartition, FetchPartitionData)]): Unit = { +// Check the topic partition only since we are reusing this callback on different TopicIdPartitions. +successfulFetch = response.headOption.filter { case (topicIdPartition, _) => topicIdPartition.topicPartition == tidp.topicPartition }.map { case (_, data) => data } Review comment: STILL TODO for Friday -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [kafka] jolshan commented on a change in pull request #11331: KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs
jolshan commented on a change in pull request #11331: URL: https://github.com/apache/kafka/pull/11331#discussion_r743370521 ## File path: core/src/test/scala/unit/kafka/server/FetchSessionTest.scala ## @@ -659,88 +670,108 @@ class FetchSessionTest { } @Test - def testIncrementalFetchSessionWithIdsWhenSessionDoesNotUseIds() : Unit = { + def testFetchSessionWithUnknownId(): Unit = { val time = new MockTime() val cache = new FetchSessionCache(10, 1000) val fetchManager = new FetchManager(time, cache) -val topicIds = new util.HashMap[String, Uuid]() -val topicNames = new util.HashMap[Uuid, String]() +val topicNames = Map(Uuid.randomUuid() -> "foo", Uuid.randomUuid() -> "bar").asJava Review comment: I can think more on this. -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Commented] (KAFKA-13299) Accept listeners that have the same port but use IPv4 vs IPv6
[ https://issues.apache.org/jira/browse/KAFKA-13299?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17439054#comment-17439054 ] Luke Chen commented on KAFKA-13299: --- [~mdedetrich-aiven], thanks for reporting this issue. I think your suggestion make sense to me. Would you mind if I take this ticket and work on the fix? Or you're planing to work on it by yourselves? Thanks. > Accept listeners that have the same port but use IPv4 vs IPv6 > - > > Key: KAFKA-13299 > URL: https://issues.apache.org/jira/browse/KAFKA-13299 > Project: Kafka > Issue Type: Improvement >Reporter: Matthew de Detrich >Priority: Major > > Currently we are going through a process where we want to migrate Kafka > brokers from IPv4 to IPv6. The simplest way for us to do this would be to > allow Kafka to have 2 listeners of the same port however one listener has an > IPv4 address allocated and another listener has an IPv6 address allocated. > Currently this is not possible in Kafka because it validates that all of the > listeners have a unique port. With some rudimentary testing if this > validation is removed (so we are able to have 2 listeners of the same port > but with different IP versions) there doesn't seem to be any immediate > problems, the kafka clusters works without any problems. > Is there some fundamental reason behind this limitation of having unique > ports? Consequently would there be any problems in loosening this limitation > (i.e. duplicate ports are allowed if the IP versions are different) or just > altogether removing the restriction -- This message was sent by Atlassian Jira (v8.3.4#803005)