Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Remove ConsumerRebalanceListener from ConsumerVerticleContext #3207

Merged
merged 3 commits into from
Jul 13, 2023
Merged

Remove ConsumerRebalanceListener from ConsumerVerticleContext #3207

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c4d8cbb
Implemented
Jul 11, 2023
8b2d531
Test fixed acording
Jul 11, 2023
7ace30c
Change the Tests to set RebalanceListener
Jul 12, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
10 changes: 10 additions & 0 deletions ...ain/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/ConsumerVerticle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import io.vertx.core.Vertx;
import java.util.Objects;
import java.util.function.BiFunction;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -40,6 +41,7 @@ public interface Initializer extends BiFunction<Vertx, ConsumerVerticle, Future<
private final ConsumerVerticleContext consumerVerticleContext;

ReactiveKafkaConsumer<Object, CloudEvent> consumer;
ConsumerRebalanceListener consumerRebalanceListener;
RecordDispatcher recordDispatcher;
private AsyncCloseable closeable;

Expand Down Expand Up @@ -85,6 +87,10 @@ public void setCloser(AsyncCloseable closeable) {
this.closeable = closeable;
}

public void setRebalanceListener(ConsumerRebalanceListener consumerRebalanceListener) {
this.consumerRebalanceListener = consumerRebalanceListener;
}

void exceptionHandler(Throwable cause) {
logger.error("Consumer exception {}", consumerVerticleContext.getLoggingKeyValue(), cause);

Expand All @@ -98,5 +104,9 @@ protected ConsumerVerticleContext getConsumerVerticleContext() {
return consumerVerticleContext;
}

protected ConsumerRebalanceListener getConsumerRebalanceListener() {
return consumerRebalanceListener;
}

public abstract PartitionRevokedHandler getPartitionRevokedHandler();
}
4 changes: 2 additions & 2 deletions ...a/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,12 +94,12 @@ public OrderedConsumerVerticle(final ConsumerVerticleContext context, final Init

@Override
void startConsumer(Promise<Void> startPromise) {
Objects.requireNonNull(getConsumerVerticleContext().getConsumerRebalanceListener());
Objects.requireNonNull(getConsumerRebalanceListener());
// We need to sub first, then we can start the polling loop
this.consumer
.subscribe(
Set.copyOf(getConsumerVerticleContext().getResource().getTopicsList()),
getConsumerVerticleContext().getConsumerRebalanceListener())
getConsumerRebalanceListener())
.onFailure(startPromise::fail)
.onSuccess(v -> {
if (this.pollTimer.compareAndSet(-1, 0)) {
Expand Down
4 changes: 2 additions & 2 deletions ...dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,12 +57,12 @@ public UnorderedConsumerVerticle(final ConsumerVerticleContext context, final In

@Override
void startConsumer(final Promise<Void> startPromise) {
Objects.requireNonNull(getConsumerVerticleContext().getConsumerRebalanceListener());
Objects.requireNonNull(getConsumerRebalanceListener());

this.consumer
.subscribe(
Set.copyOf(getConsumerVerticleContext().getResource().getTopicsList()),
getConsumerVerticleContext().getConsumerRebalanceListener())
getConsumerRebalanceListener())
.onFailure(startPromise::tryFail)
.onSuccess(startPromise::tryComplete)
.onSuccess(v -> poll());
Expand Down
2 changes: 1 addition & 1 deletion .../main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ private void build(final Vertx vertx, final ConsumerVerticle consumerVerticle, f

final var partitionRevokedHandlers =
List.of(consumerVerticle.getPartitionRevokedHandler(), offsetManager.getPartitionRevokedHandler());
this.consumerVerticleContext.withConsumerRebalanceListener(createRebalanceListener(partitionRevokedHandlers));
consumerVerticle.setRebalanceListener(createRebalanceListener(partitionRevokedHandlers));
}

private ConsumerVerticle createConsumerVerticle(final ConsumerVerticle.Initializer initializer) {
Expand Down
13 changes: 0 additions & 13 deletions .../main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,6 @@
import java.util.Map;
import java.util.Objects;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -64,8 +63,6 @@ public class ConsumerVerticleContext {

private Tags tags;

private ConsumerRebalanceListener consumerRebalanceListener;

public ConsumerVerticleContext withConsumerConfigs(final Map<String, Object> consumerConfigs) {
this.consumerConfigs = new HashMap<>(consumerConfigs);
return this;
Expand Down Expand Up @@ -157,16 +154,6 @@ public ConsumerVerticleContext withProducerFactory(final ProducerFactory<String,
return this;
}

public ConsumerVerticleContext withConsumerRebalanceListener(
final ConsumerRebalanceListener consumerRebalanceListener) {
this.consumerRebalanceListener = consumerRebalanceListener;
return this;
}

public ConsumerRebalanceListener getConsumerRebalanceListener() {
return consumerRebalanceListener;
}

public DataPlaneContract.Resource getResource() {
return resource;
}
Expand Down
20 changes: 20 additions & 0 deletions .../knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
import io.vertx.micrometer.backends.BackendRegistries;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
Expand Down Expand Up @@ -89,7 +90,13 @@ public void subscribedToTopic(final Vertx vertx, final VertxTestContext context)
consumerVerticle.setConsumer(new MockReactiveKafkaConsumer<>(consumer));
consumerVerticle.setRecordDispatcher(recordDispatcher);
consumerVerticle.setCloser(Future::succeededFuture);
consumerVerticle.setRebalanceListener(new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {}

@Override
public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {}
});
return Future.succeededFuture();
});

Expand Down Expand Up @@ -130,6 +137,13 @@ public void stop(final Vertx vertx, final VertxTestContext context) {
consumerVerticle.setConsumer(new MockReactiveKafkaConsumer<>(consumer));
consumerVerticle.setRecordDispatcher(recordDispatcher);
consumerVerticle.setCloser(Future::succeededFuture);
consumerVerticle.setRebalanceListener(new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {}

@Override
public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {}
});

return Future.succeededFuture();
});
Expand Down Expand Up @@ -207,7 +221,13 @@ public void shouldCloseEverything(final Vertx vertx, final VertxTestContext cont
consumerVerticle.setConsumer(consumer);
consumerVerticle.setRecordDispatcher(recordDispatcher);
consumerVerticle.setCloser(Future::succeededFuture);
consumerVerticle.setRebalanceListener(new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {}

@Override
public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {}
});
return Future.succeededFuture();
});

Expand Down
8 changes: 8 additions & 0 deletions ...v/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticleTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import io.vertx.core.Vertx;
import io.vertx.junit5.VertxExtension;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
Expand All @@ -43,6 +44,7 @@
import java.util.stream.IntStream;
import java.util.stream.LongStream;
import java.util.stream.Stream;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
Expand Down Expand Up @@ -123,7 +125,13 @@ public void consumeOneByOne(
consumerVerticle.setConsumer(new MockReactiveKafkaConsumer<>(consumer));
consumerVerticle.setRecordDispatcher(recordDispatcher);
consumerVerticle.setCloser(Future::succeededFuture);
consumerVerticle.setRebalanceListener(new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {}

@Override
public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {}
});
return Future.succeededFuture();
});

Expand Down
23 changes: 2 additions & 21 deletions ...t/java/dev/knative/eventing/kafka/broker/dispatcher/main/FakeConsumerVerticleContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import dev.knative.eventing.kafka.broker.core.testing.CoreObjects;
import io.vertx.ext.web.client.WebClientOptions;
import java.util.HashMap;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;

public class FakeConsumerVerticleContext {

Expand All @@ -30,16 +29,7 @@ public static ConsumerVerticleContext get() {
.withProducerConfigs(new HashMap<>())
.withConsumerConfigs(new HashMap<>())
.withMeterRegistry(Metrics.getRegistry())
.withResource(CoreObjects.resource1(), CoreObjects.egress1())
.withConsumerRebalanceListener(new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(
final java.util.Collection<org.apache.kafka.common.TopicPartition> partitions) {}

@Override
public void onPartitionsAssigned(
final java.util.Collection<org.apache.kafka.common.TopicPartition> partitions) {}
});
.withResource(CoreObjects.resource1(), CoreObjects.egress1());
}

public static ConsumerVerticleContext get(
Expand All @@ -50,15 +40,6 @@ public static ConsumerVerticleContext get(
.withAuthProvider(AuthProvider.noAuth())
.withWebClientOptions(new WebClientOptions())
.withMeterRegistry(Metrics.getRegistry())
.withResource(resource, egress)
.withConsumerRebalanceListener(new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(
final java.util.Collection<org.apache.kafka.common.TopicPartition> partitions) {}

@Override
public void onPartitionsAssigned(
final java.util.Collection<org.apache.kafka.common.TopicPartition> partitions) {}
});
.withResource(resource, egress);
}
}