chia7712 commented on code in PR #22271:
URL: https://github.com/apache/kafka/pull/22271#discussion_r3695896686
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCallbackTest.java:
##########
@@ -89,7 +90,7 @@ public void
testAsyncConsumerRebalanceListenerAssignmentOnPartitionsAssigned() t
private void
testRebalanceListenerAssignmentOnPartitionsAssigned(GroupProtocol
groupProtocol) throws InterruptedException {
try (var consumer = createConsumer(groupProtocol)) {
triggerOnPartitionsAssigned(tp, consumer,
- (executeConsumer, partitions) ->
assertTrue(executeConsumer.assignment().contains(tp))
+ (executeConsumer, partitions) ->
assertTrue(executeConsumer.assignment().contains(tp))
Review Comment:
Please revert unrelated change
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerRebalanceListenerInvoker.java:
##########
@@ -86,19 +81,16 @@ public Exception invokePartitionsRevoked(final
SortedSet<TopicPartition> revoked
if (!revokePausedPartitions.isEmpty())
log.info("The pause flag in partitions {} will be removed due to
revocation.", revokePausedPartitions);
- Optional<ConsumerRebalanceListener> listener =
subscriptions.rebalanceListener();
-
- if (listener.isPresent()) {
+ if (subscriptions.hasRebalanceListener()) {
try {
final long startMs = time.milliseconds();
- listener.get().onPartitionsRevoked(revokedPartitions);
+ subscriptions.onPartitionsRevoked(revokedPartitions);
metricsManager.recordPartitionsRevokedLatency(time.milliseconds() - startMs);
} catch (WakeupException | InterruptException e) {
throw e;
} catch (Exception e) {
log.error(
- "User provided listener {} failed on invocation of
onPartitionsRevoked for partitions {}",
- listener.get().getClass().getName(),
Review Comment:
ditto
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCallbackTest.java:
##########
@@ -134,14 +135,14 @@ private void
testRebalanceListenerAssignOnPartitionsRevoked(GroupProtocol groupP
@ClusterTest
public void
testClassicConsumerRebalanceListenerAssignmentOnPartitionsRevoked() throws
InterruptedException {
triggerOnPartitionsRevoked(tp, CLASSIC,
- (consumer, partitions) ->
assertTrue(consumer.assignment().contains(tp))
+ (consumer, partitions) ->
assertTrue(consumer.assignment().contains(tp))
);
}
@ClusterTest
public void
testAsyncConsumerRebalanceListenerAssignmentOnPartitionsRevoked() throws
InterruptedException {
triggerOnPartitionsRevoked(tp, CONSUMER,
- (consumer, partitions) ->
assertTrue(consumer.assignment().contains(tp))
+ (consumer, partitions) ->
assertTrue(consumer.assignment().contains(tp))
Review Comment:
ditto
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerRebalanceListenerInvoker.java:
##########
@@ -116,19 +108,16 @@ public Exception invokePartitionsLost(final
SortedSet<TopicPartition> lostPartit
if (!lostPausedPartitions.isEmpty())
log.info("The pause flag in partitions {} will be removed due to
partition lost.", lostPartitions);
- Optional<ConsumerRebalanceListener> listener =
subscriptions.rebalanceListener();
-
- if (listener.isPresent()) {
+ if (subscriptions.hasRebalanceListener()) {
try {
final long startMs = time.milliseconds();
- listener.get().onPartitionsLost(lostPartitions);
+ subscriptions.onPartitionsLost(lostPartitions);
metricsManager.recordPartitionsLostLatency(time.milliseconds()
- startMs);
} catch (WakeupException | InterruptException e) {
throw e;
} catch (Exception e) {
log.error(
- "User provided listener {} failed on invocation of
onPartitionsLost for partitions {}",
- listener.get().getClass().getName(),
Review Comment:
ditto
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerRebalanceListenerInvoker.java:
##########
@@ -56,19 +54,16 @@ public class ConsumerRebalanceListenerInvoker {
public Exception invokePartitionsAssigned(final SortedSet<TopicPartition>
assignedPartitions) {
log.info("Adding newly assigned partitions: {}", assignedPartitions);
- Optional<ConsumerRebalanceListener> listener =
subscriptions.rebalanceListener();
-
- if (listener.isPresent()) {
+ if (subscriptions.hasRebalanceListener()) {
try {
final long startMs = time.milliseconds();
- listener.get().onPartitionsAssigned(assignedPartitions);
+ subscriptions.onPartitionsAssigned(assignedPartitions);
metricsManager.recordPartitionsAssignedLatency(time.milliseconds() - startMs);
} catch (WakeupException | InterruptException e) {
throw e;
} catch (Exception e) {
log.error(
- "User provided listener {} failed on invocation of
onPartitionsAssigned for partitions {}",
- listener.get().getClass().getName(),
Review Comment:
Could you keep this log meesage?
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCallbackTest.java:
##########
@@ -134,14 +135,14 @@ private void
testRebalanceListenerAssignOnPartitionsRevoked(GroupProtocol groupP
@ClusterTest
public void
testClassicConsumerRebalanceListenerAssignmentOnPartitionsRevoked() throws
InterruptedException {
triggerOnPartitionsRevoked(tp, CLASSIC,
- (consumer, partitions) ->
assertTrue(consumer.assignment().contains(tp))
+ (consumer, partitions) ->
assertTrue(consumer.assignment().contains(tp))
Review Comment:
ditto
--
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]
