ShivsundarR commented on code in PR #18105:
URL: https://github.com/apache/kafka/pull/18105#discussion_r1878005106
##########
core/src/test/java/kafka/test/api/ShareConsumerTest.java:
##########
@@ -1843,47 +1725,47 @@ private void produceMessagesWithTimestamp(int
messageCount, long startingTimesta
}
}
- private void consumeMessages(AtomicInteger totalMessagesConsumed,
+ private int consumeMessages(AtomicInteger totalMessagesConsumed,
int totalMessages,
String groupId,
int consumerNumber,
int maxPolls,
- boolean commit,
- CompletableFuture<Integer> future) {
- try (KafkaShareConsumer<byte[], byte[]> shareConsumer =
createShareConsumer(new ByteArrayDeserializer(), new ByteArrayDeserializer(),
groupId)) {
- shareConsumer.subscribe(Collections.singleton(tp.topic()));
- consumeMessages(shareConsumer, totalMessagesConsumed,
totalMessages, consumerNumber, maxPolls, commit, future);
- } catch (Exception e) {
- fail("Consumer " + consumerNumber + " failed with exception: " +
e);
- }
+ boolean commit) {
+ return assertDoesNotThrow(() -> {
+ try (KafkaShareConsumer<byte[], byte[]> shareConsumer =
createShareConsumer(
+ new ByteArrayDeserializer(), new ByteArrayDeserializer(),
groupId)) {
+ shareConsumer.subscribe(Collections.singleton(tp.topic()));
+ return consumeMessages(shareConsumer, totalMessagesConsumed,
totalMessages, consumerNumber, maxPolls, commit);
Review Comment:
Here as well, can we add the custom error message with consumer number as
well.
##########
core/src/test/java/kafka/test/api/ShareConsumerTest.java:
##########
@@ -1843,47 +1725,47 @@ private void produceMessagesWithTimestamp(int
messageCount, long startingTimesta
}
}
- private void consumeMessages(AtomicInteger totalMessagesConsumed,
+ private int consumeMessages(AtomicInteger totalMessagesConsumed,
int totalMessages,
String groupId,
int consumerNumber,
int maxPolls,
- boolean commit,
- CompletableFuture<Integer> future) {
- try (KafkaShareConsumer<byte[], byte[]> shareConsumer =
createShareConsumer(new ByteArrayDeserializer(), new ByteArrayDeserializer(),
groupId)) {
- shareConsumer.subscribe(Collections.singleton(tp.topic()));
- consumeMessages(shareConsumer, totalMessagesConsumed,
totalMessages, consumerNumber, maxPolls, commit, future);
- } catch (Exception e) {
- fail("Consumer " + consumerNumber + " failed with exception: " +
e);
- }
+ boolean commit) {
+ return assertDoesNotThrow(() -> {
+ try (KafkaShareConsumer<byte[], byte[]> shareConsumer =
createShareConsumer(
+ new ByteArrayDeserializer(), new ByteArrayDeserializer(),
groupId)) {
+ shareConsumer.subscribe(Collections.singleton(tp.topic()));
+ return consumeMessages(shareConsumer, totalMessagesConsumed,
totalMessages, consumerNumber, maxPolls, commit);
Review Comment:
Here as well, can we add the custom error message with consumer number.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
