You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "gongxuanzhang (via GitHub)" <gi...@apache.org> on 2024/06/11 06:49:05 UTC
[PR] [MINOR] Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
gongxuanzhang opened a new pull request, #16277:
URL: https://github.com/apache/kafka/pull/16277
The original author added a note in `BuiltInPartitioner`
> Visible and used for testing only
I think this will break the scope.
After compilation, the test module and the source code are in the same package.
So we don't need `public`, just need `protected`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #16277:
URL: https://github.com/apache/kafka/pull/16277
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16277:
URL: https://github.com/apache/kafka/pull/16277#discussion_r1634581946
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java:
##########
@@ -45,7 +45,7 @@ public class BuiltInPartitioner {
private final AtomicReference<StickyPartitionInfo> stickyPartitionInfo = new AtomicReference<>();
// Visible and used for testing only.
- static volatile public Supplier<Integer> mockRandom = null;
+ protected static volatile Supplier<Integer> mockRandom = null;
Review Comment:
Agree that could introduce some changes, but we can minimize the changes by opening only required methods. For example:
**BuiltInPartitioner**
```java
int random() {
return Utils.toPositive(ThreadLocalRandom.current().nextInt());
}
private int nextPartition(Cluster cluster) {
int random = random();
...
}
```
**RecordAccumulator**
```java
private Deque<ProducerBatch> getOrCreateDeque(TopicPartition tp) {
TopicInfo topicInfo = topicInfoMap.computeIfAbsent(tp.topic(), k -> new TopicInfo(createBuiltInPartitioner(logContext, k, batchSize)));
return topicInfo.batches.computeIfAbsent(tp.partition(), k -> new ArrayDeque<>());
}
BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic, int stickyBatchSize) {
return new BuiltInPartitioner(logContext, topic, stickyBatchSize);
}
public TopicInfo(BuiltInPartitioner builtInPartitioner) {
this.builtInPartitioner = builtInPartitioner;
}
```
With those override-able methods, both `RecordAccumulatorTest` and `BuiltInPartitionerTest` can create `BuiltInPartitioner` with custom `random` method. 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #16277:
URL: https://github.com/apache/kafka/pull/16277#discussion_r1634566222
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java:
##########
@@ -45,7 +45,7 @@ public class BuiltInPartitioner {
private final AtomicReference<StickyPartitionInfo> stickyPartitionInfo = new AtomicReference<>();
// Visible and used for testing only.
- static volatile public Supplier<Integer> mockRandom = null;
+ protected static volatile Supplier<Integer> mockRandom = null;
Review Comment:
`mockRandom` is a static field, which means its behavior can be modified anywhere in the test. There are many instances in the context, not just within the test packages. Therefore, if we add a function and let the test module extend this class, it will cause the already created objects to not work as expected, requiring significant code changes.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #16277:
URL: https://github.com/apache/kafka/pull/16277#discussion_r1635719559
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -1763,4 +1747,65 @@ private RecordAccumulator createTestRecordAccumulator(
txnManager,
new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
}
+
+ private class SequentialRecordAccumulator extends RecordAccumulator {
Review Comment:
This is an excellent suggestion!, perfectly reproducing the "global modification" in the previous code.
I changed PR, plz review it @chia7712
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #16277:
URL: https://github.com/apache/kafka/pull/16277#discussion_r1637379530
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -1033,10 +1033,18 @@ public Deque<ProducerBatch> getDeque(TopicPartition tp) {
* Get the deque for the given topic-partition, creating it if necessary.
*/
private Deque<ProducerBatch> getOrCreateDeque(TopicPartition tp) {
- TopicInfo topicInfo = topicInfoMap.computeIfAbsent(tp.topic(), k -> new TopicInfo(logContext, k, batchSize));
+ TopicInfo topicInfo = topicInfoMap.computeIfAbsent(tp.topic(),
+ k -> new TopicInfo(createBuiltInPartitioner(logContext, k, batchSize)));
return topicInfo.batches.computeIfAbsent(tp.partition(), k -> new ArrayDeque<>());
}
+ /**
+ * Subclass can custom {@link BuiltInPartitioner}
Review Comment:
You are right,it seem to don’t need explaining. I removed docs
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #16277:
URL: https://github.com/apache/kafka/pull/16277#discussion_r1634859772
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java:
##########
@@ -45,7 +45,7 @@ public class BuiltInPartitioner {
private final AtomicReference<StickyPartitionInfo> stickyPartitionInfo = new AtomicReference<>();
// Visible and used for testing only.
- static volatile public Supplier<Integer> mockRandom = null;
+ protected static volatile Supplier<Integer> mockRandom = null;
Review Comment:
I modified this PR.
BTW,I moved the hard-coded parts to a private class within the test module in order to pass the original tests
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16277:
URL: https://github.com/apache/kafka/pull/16277#discussion_r1634421267
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java:
##########
@@ -45,7 +45,7 @@ public class BuiltInPartitioner {
private final AtomicReference<StickyPartitionInfo> stickyPartitionInfo = new AtomicReference<>();
// Visible and used for testing only.
- static volatile public Supplier<Integer> mockRandom = null;
+ protected static volatile Supplier<Integer> mockRandom = null;
Review Comment:
Maybe a better way is to add a package-private method `random()`, and then in the `BuiltInPartitionerTest` we create a `BuiltInPartitioner` with override `random()`. With that change we can remove "mock" from `BuiltInPartitioner`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16277:
URL: https://github.com/apache/kafka/pull/16277#discussion_r1635142024
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -1763,4 +1747,65 @@ private RecordAccumulator createTestRecordAccumulator(
txnManager,
new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
}
+
+ private class SequentialRecordAccumulator extends RecordAccumulator {
Review Comment:
Could we move `AtomicInteger mockRandom` to class-level and then return `RecordAccumulator` with override `createBuiltInPartitioner`? For example:
**RecordAccumulatorTest**
```java
private AtomicInteger random = null;
@AfterEach
public void teardown() {
this.metrics.close();
}
private RecordAccumulator createTestRecordAccumulator(
TransactionManager txnManager,
int deliveryTimeoutMs,
int batchSize,
long totalSize,
Compression compression,
int lingerMs
) {
long retryBackoffMs = 100L;
long retryBackoffMaxMs = 1000L;
String metricGrpName = "producer-metrics";
return new RecordAccumulator(
logContext,
batchSize,
compression,
lingerMs,
retryBackoffMs,
retryBackoffMaxMs,
deliveryTimeoutMs,
metrics,
metricGrpName,
time,
new ApiVersions(),
txnManager,
new BufferPool(totalSize, batchSize, metrics, time, metricGrpName)) {
@Override
BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic, int stickyBatchSize) {
return new BuiltInPartitioner(logContext, topic, stickyBatchSize) {
@Override
int random() {
return random == null ? super.random() : random.getAndIncrement();
}
};
}
};
}
```
With that change, `testUniformBuiltInPartitioner` can define the `random` in the begining
```java
random = new AtomicInteger();
```
`testAdaptiveBuiltInPartitioner` can create `RecordAccumulator` with override `createBuiltInPartitioner` directly.
```java
RecordAccumulator accum = new RecordAccumulator(logContext, batchSize, Compression.NONE, 0, 0L, 0L,
3200, config, metrics, "producer-metrics", time, new ApiVersions(), null,
new BufferPool(totalSize, batchSize, metrics, time, "producer-internal-metrics")) {
@Override
BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic, int stickyBatchSize) {
return new BuiltInPartitioner(logContext, topic, stickyBatchSize) {
@Override
int random() {
return random == null ? super.random() : random.getAndIncrement();
}
};
}
};
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16277:
URL: https://github.com/apache/kafka/pull/16277#discussion_r1635740247
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -103,13 +103,13 @@ public class RecordAccumulatorTest {
private final Map<Integer, Node> nodes = Stream.of(node1, node2).collect(Collectors.toMap(Node::id, Function.identity()));
private MetadataSnapshot metadataCache = new MetadataSnapshot(null,
- nodes,
- partMetadatas,
- Collections.emptySet(),
- Collections.emptySet(),
- Collections.emptySet(),
- null,
- Collections.emptyMap());
+ nodes,
+ partMetadatas,
Review Comment:
please revert unrelated changes
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -121,12 +121,14 @@ public class RecordAccumulatorTest {
private final Metrics metrics = new Metrics(time);
private final long maxBlockTimeMs = 1000;
private final LogContext logContext = new LogContext();
+ private AtomicInteger mockRandom = null;
@BeforeEach void setup() {}
@AfterEach
public void teardown() {
this.metrics.close();
+ mockRandom = null;
Review Comment:
this is redundant since the variable is case-level
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -146,17 +148,17 @@ public void testDrainBatches() throws Exception {
PartitionInfo part3 = MetadataResponse.toPartitionInfo(partMetadata3, nodes);
PartitionInfo part4 = MetadataResponse.toPartitionInfo(partMetadata4, nodes);
Cluster cluster = new Cluster(null, Arrays.asList(node1, node2), Arrays.asList(part1, part2, part3, part4),
- Collections.emptySet(), Collections.emptySet());
+ Collections.emptySet(), Collections.emptySet());
metadataCache = new MetadataSnapshot(null,
- nodes,
- partMetadatas,
- Collections.emptySet(),
- Collections.emptySet(),
- Collections.emptySet(),
- null,
- Collections.emptyMap(),
- cluster);
+ nodes,
+ partMetadatas,
+ Collections.emptySet(),
+ Collections.emptySet(),
Review Comment:
ditto
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -1737,30 +1735,48 @@ private RecordAccumulator createTestRecordAccumulator(int deliveryTimeoutMs, int
* Return a test RecordAccumulator instance
*/
private RecordAccumulator createTestRecordAccumulator(
- TransactionManager txnManager,
- int deliveryTimeoutMs,
- int batchSize,
- long totalSize,
- Compression compression,
- int lingerMs
+ TransactionManager txnManager,
+ int deliveryTimeoutMs,
+ int batchSize,
+ long totalSize,
+ Compression compression,
+ int lingerMs
) {
long retryBackoffMs = 100L;
long retryBackoffMaxMs = 1000L;
String metricGrpName = "producer-metrics";
return new RecordAccumulator(
- logContext,
- batchSize,
- compression,
- lingerMs,
- retryBackoffMs,
- retryBackoffMaxMs,
- deliveryTimeoutMs,
- metrics,
- metricGrpName,
- time,
- new ApiVersions(),
- txnManager,
- new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
+ logContext,
+ batchSize,
+ compression,
+ lingerMs,
+ retryBackoffMs,
+ retryBackoffMaxMs,
+ deliveryTimeoutMs,
+ metrics,
+ metricGrpName,
+ time,
+ new ApiVersions(),
+ txnManager,
+ new BufferPool(totalSize, batchSize, metrics, time, metricGrpName)) {
+ @Override
+ protected BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic,
+ int stickyBatchSize) {
+ return new MockRandomBuiltInPartitioner(logContext, topic, stickyBatchSize);
+ }
+ };
+ }
+
+ private class MockRandomBuiltInPartitioner extends BuiltInPartitioner {
Review Comment:
it can be static class, and it would be great to unify the naming -> `SequentialPartitioner`
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -146,17 +148,17 @@ public void testDrainBatches() throws Exception {
PartitionInfo part3 = MetadataResponse.toPartitionInfo(partMetadata3, nodes);
PartitionInfo part4 = MetadataResponse.toPartitionInfo(partMetadata4, nodes);
Cluster cluster = new Cluster(null, Arrays.asList(node1, node2), Arrays.asList(part1, part2, part3, part4),
- Collections.emptySet(), Collections.emptySet());
+ Collections.emptySet(), Collections.emptySet());
Review Comment:
ditto
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/BuiltInPartitionerTest.java:
##########
@@ -203,4 +189,19 @@ void testStickyBatchSizeMoreThatZero() {
assertThrows(IllegalArgumentException.class, () -> new BuiltInPartitioner(logContext, TOPIC_A, 0));
assertDoesNotThrow(() -> new BuiltInPartitioner(logContext, TOPIC_A, 1));
}
+
+
+ static class SequentialPartitioner extends BuiltInPartitioner {
Review Comment:
could you please make it be private ?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -1033,9 +1033,17 @@ public Deque<ProducerBatch> getDeque(TopicPartition tp) {
* Get the deque for the given topic-partition, creating it if necessary.
*/
private Deque<ProducerBatch> getOrCreateDeque(TopicPartition tp) {
- TopicInfo topicInfo = topicInfoMap.computeIfAbsent(tp.topic(), k -> new TopicInfo(logContext, k, batchSize));
+ TopicInfo topicInfo = topicInfoMap.computeIfAbsent(tp.topic(),
+ k -> new TopicInfo(createBuiltInPartitioner(logContext, k, batchSize)));
return topicInfo.batches.computeIfAbsent(tp.partition(), k -> new ArrayDeque<>());
}
+
+ /**
+ * Subclass can custom {@link BuiltInPartitioner}
+ **/
+ protected BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic, int stickyBatchSize) {
Review Comment:
Could you change it to package-private and add visible for testing?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java:
##########
@@ -113,6 +110,10 @@ private int nextPartition(Cluster cluster) {
return partition;
}
+ protected int randomPartition() {
Review Comment:
Could you change it to package-private and add `visible for testing`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on code in PR #16277:
URL: https://github.com/apache/kafka/pull/16277#discussion_r1635913417
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -1737,30 +1735,48 @@ private RecordAccumulator createTestRecordAccumulator(int deliveryTimeoutMs, int
* Return a test RecordAccumulator instance
*/
private RecordAccumulator createTestRecordAccumulator(
- TransactionManager txnManager,
- int deliveryTimeoutMs,
- int batchSize,
- long totalSize,
- Compression compression,
- int lingerMs
+ TransactionManager txnManager,
+ int deliveryTimeoutMs,
+ int batchSize,
+ long totalSize,
+ Compression compression,
+ int lingerMs
) {
long retryBackoffMs = 100L;
long retryBackoffMaxMs = 1000L;
String metricGrpName = "producer-metrics";
return new RecordAccumulator(
- logContext,
- batchSize,
- compression,
- lingerMs,
- retryBackoffMs,
- retryBackoffMaxMs,
- deliveryTimeoutMs,
- metrics,
- metricGrpName,
- time,
- new ApiVersions(),
- txnManager,
- new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
+ logContext,
+ batchSize,
+ compression,
+ lingerMs,
+ retryBackoffMs,
+ retryBackoffMaxMs,
+ deliveryTimeoutMs,
+ metrics,
+ metricGrpName,
+ time,
+ new ApiVersions(),
+ txnManager,
+ new BufferPool(totalSize, batchSize, metrics, time, metricGrpName)) {
+ @Override
+ protected BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic,
+ int stickyBatchSize) {
+ return new MockRandomBuiltInPartitioner(logContext, topic, stickyBatchSize);
+ }
+ };
+ }
+
+ private class MockRandomBuiltInPartitioner extends BuiltInPartitioner {
Review Comment:
It needs to access the instance properties of the test class so can't be static class
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16277:
URL: https://github.com/apache/kafka/pull/16277#discussion_r1635920882
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java:
##########
@@ -1737,30 +1735,48 @@ private RecordAccumulator createTestRecordAccumulator(int deliveryTimeoutMs, int
* Return a test RecordAccumulator instance
*/
private RecordAccumulator createTestRecordAccumulator(
- TransactionManager txnManager,
- int deliveryTimeoutMs,
- int batchSize,
- long totalSize,
- Compression compression,
- int lingerMs
+ TransactionManager txnManager,
+ int deliveryTimeoutMs,
+ int batchSize,
+ long totalSize,
+ Compression compression,
+ int lingerMs
) {
long retryBackoffMs = 100L;
long retryBackoffMaxMs = 1000L;
String metricGrpName = "producer-metrics";
return new RecordAccumulator(
- logContext,
- batchSize,
- compression,
- lingerMs,
- retryBackoffMs,
- retryBackoffMaxMs,
- deliveryTimeoutMs,
- metrics,
- metricGrpName,
- time,
- new ApiVersions(),
- txnManager,
- new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
+ logContext,
+ batchSize,
+ compression,
+ lingerMs,
+ retryBackoffMs,
+ retryBackoffMaxMs,
+ deliveryTimeoutMs,
+ metrics,
+ metricGrpName,
+ time,
+ new ApiVersions(),
+ txnManager,
+ new BufferPool(totalSize, batchSize, metrics, time, metricGrpName)) {
+ @Override
+ protected BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic,
+ int stickyBatchSize) {
+ return new MockRandomBuiltInPartitioner(logContext, topic, stickyBatchSize);
+ }
+ };
+ }
+
+ private class MockRandomBuiltInPartitioner extends BuiltInPartitioner {
Review Comment:
Got it. thanks!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "gongxuanzhang (via GitHub)" <gi...@apache.org>.
gongxuanzhang commented on PR #16277:
URL: https://github.com/apache/kafka/pull/16277#issuecomment-2162285665
I updated it ,plz review 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [MINOR]:Update visibility from public to protected and adjust the order in BuiltInPartitioner [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16277:
URL: https://github.com/apache/kafka/pull/16277#discussion_r1637123570
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -1033,10 +1033,18 @@ public Deque<ProducerBatch> getDeque(TopicPartition tp) {
* Get the deque for the given topic-partition, creating it if necessary.
*/
private Deque<ProducerBatch> getOrCreateDeque(TopicPartition tp) {
- TopicInfo topicInfo = topicInfoMap.computeIfAbsent(tp.topic(), k -> new TopicInfo(logContext, k, batchSize));
+ TopicInfo topicInfo = topicInfoMap.computeIfAbsent(tp.topic(),
+ k -> new TopicInfo(createBuiltInPartitioner(logContext, k, batchSize)));
return topicInfo.batches.computeIfAbsent(tp.partition(), k -> new ArrayDeque<>());
}
+ /**
+ * Subclass can custom {@link BuiltInPartitioner}
Review Comment:
this comment is a bit redundant, since it just describe what the code. It would be better to explain why we need this method rather than inline it. Or we can remove the docs
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org