This repository was archived by the owner on Jan 24, 2024. It is now read-only.
[feature] Support message publish rate limiting #1589
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
392abd3 to
2524538
Compare
Codecov Report
@@ Coverage Diff @@
## master #1589 +/- ##
============================================
- Coverage 15.34% 14.85% -0.49%
+ Complexity 592 589 -3
============================================
Files 164 164
Lines 11913 12225 +312
Branches 1102 1120 +18
============================================
- Hits 1828 1816 -12
- Misses 9931 10254 +323
- Partials 154 155 +1
|
426e704 to
9382ee3
Compare
Member
Author
|
@BewareMyPower @eolivelli Please help review this PR :-) |
Collaborator
|
I will review it next week. |
Collaborator
|
Not related to this PR, I think it would be good to implement KIP-219 |
kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLog.java
Outdated
Show resolved
Hide resolved
kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLog.java
Outdated
Show resolved
Hide resolved
tests/src/test/java/io/streamnative/pulsar/handlers/kop/MessagePublishThrottlingTest.java
Outdated
Show resolved
Hide resolved
tests/src/test/java/io/streamnative/pulsar/handlers/kop/MessagePublishThrottlingTest.java
Outdated
Show resolved
Hide resolved
Collaborator
|
We can reduce duplicated code by adding a method like: private void sendMessagesFromMultiThreads(KafkaProducer<Integer, byte[]> producer, String topicName,
int numThread, int numMessage, int msgBytes) throws Exception {
ExecutorService executorService = Executors.newFixedThreadPool(numThread);
List<CompletableFuture<Void>> futures = new ArrayList<>();
for (int n = 0; n < numThread; n++) {
final CompletableFuture<Void> future = new CompletableFuture<>();
futures.add(future);
executorService.submit(() -> {
for (int i = 0; i < numMessage; i++) {
try {
producer.send(new ProducerRecord<>(topicName, new byte[msgBytes])).get();
future.complete(null);
} catch (InterruptedException | ExecutionException e) {
future.completeExceptionally(e);
}
}
});
}
FutureUtil.waitForAll(futures).get();
}It's also a fail-fast style. In addition, I noticed the two tests use different int numMessage = 10;
int msgBytes = 110;
int numThread = 4; int numMessage = 20;
int numThread = 4;
int msgBytes = 50;Is it possible to share the same config so that we can make these fields private and simplify the method to the following signature? private void sendMessagesFromMultiThreads(KafkaProducer<Integer, byte[]> producer, String topicName)
throws Exception { |
BewareMyPower
approved these changes
Dec 8, 2022
Demogorgon314
added a commit
that referenced
this pull request
Dec 8, 2022
### Motivation Currently, KoP doesn't support publish rate limit, but it is a helpful feature for users, In this PR, it introduced the message publish rate limit, and the rate limit is reused from Pulsar, it can work with Pulsar together. ### Modifications * Support message publish rate limit. * Added units test to verify it. (cherry picked from commit 60bb8f0)
Demogorgon314
added a commit
that referenced
this pull request
Dec 19, 2022
### Motivation Currently, KoP doesn't support publish rate limit, but it is a helpful feature for users, In this PR, it introduced the message publish rate limit, and the rate limit is reused from Pulsar, it can work with Pulsar together. ### Modifications * Support message publish rate limit. * Added units test to verify it. (cherry picked from commit 60bb8f0)
michaeljmarshall
pushed a commit
to michaeljmarshall/kop
that referenced
this pull request
Jan 31, 2023
Currently, KoP doesn't support publish rate limit, but it is a helpful feature for users, In this PR, it introduced the message publish rate limit, and the rate limit is reused from Pulsar, it can work with Pulsar together. * Support message publish rate limit. * Added units test to verify it.
Demogorgon314
added a commit
that referenced
this pull request
Feb 6, 2023
### Motivation Currently, KoP doesn't support publish rate limit, but it is a helpful feature for users, In this PR, it introduced the message publish rate limit, and the rate limit is reused from Pulsar, it can work with Pulsar together. ### Modifications * Support message publish rate limit. * Added units test to verify it. (cherry picked from commit 60bb8f0)
4 tasks
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
3 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
#1129
Motivation
Currently, KoP doesn't support publish rate limit, but it is a helpful feature for users,
In this PR, it introduced the message publish rate limit, and the rate limit is reused from
Pulsar, it can work with Pulsar together.
Modifications
TODO
Need check if the KoP needs to support this KIP: https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication
Documentation
Check the box below.
Need to update docs?
doc-required(If you need help on updating docs, create a doc issue)
no-need-doc(Please explain why)
doc(If this PR contains doc changes)