Abstract Publisher/Subscriber into generic interfaces
Decouple the interface to send/receive messages to Kafka
using generic interfaces bound to the actual native implementation
classes.
The KafkaSession does not refer anymore to a native KafkaProducer
allowing other implementations to be plugged in.
This is a preparation work to introduce a REST-API client
based access to Kafka REST Proxy through parameters on
the events broker plugin config.
Change-Id: I6034915c5538e3df365a45e2f134bab50aff932f
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java
index 1be52cb..2f74b44 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java
@@ -26,7 +26,7 @@
import com.googlesource.gerrit.plugins.kafka.api.KafkaApiModule;
import com.googlesource.gerrit.plugins.kafka.publish.KafkaPublisher;
import com.googlesource.gerrit.plugins.kafka.session.KafkaProducerProvider;
-import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
class Module extends AbstractModule {
@@ -43,8 +43,7 @@
DynamicSet.bind(binder(), LifecycleListener.class).to(Manager.class);
DynamicSet.bind(binder(), EventListener.class).to(KafkaPublisher.class);
- bind(new TypeLiteral<KafkaProducer<String, String>>() {})
- .toProvider(KafkaProducerProvider.class);
+ bind(new TypeLiteral<Producer<String, String>>() {}).toProvider(KafkaProducerProvider.class);
install(kafkaBrokerModule);
}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java
index 73c7509..b76ff7f 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java
@@ -28,6 +28,8 @@
import com.googlesource.gerrit.plugins.kafka.broker.ConsumerExecutor;
import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
import com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventDeserializer;
+import com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventNativeSubscriber;
+import com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
@@ -54,6 +56,7 @@
@Override
protected void configure() {
+ bind(KafkaEventSubscriber.class).to(KafkaEventNativeSubscriber.class);
bind(ExecutorService.class)
.annotatedWith(ConsumerExecutor.class)
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaProducerProvider.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaProducerProvider.java
index b1f11f7..4fb98b2 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaProducerProvider.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaProducerProvider.java
@@ -18,8 +18,9 @@
import com.google.inject.Provider;
import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
-public class KafkaProducerProvider implements Provider<KafkaProducer<String, String>> {
+public class KafkaProducerProvider implements Provider<Producer<String, String>> {
private final KafkaProperties properties;
@Inject
@@ -28,7 +29,7 @@
}
@Override
- public KafkaProducer<String, String> get() {
+ public Producer<String, String> get() {
return new KafkaProducer<>(properties);
}
}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java
index 0e851aa..f7105f9 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java
@@ -19,7 +19,6 @@
import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
import com.googlesource.gerrit.plugins.kafka.publish.KafkaEventsPublisherMetrics;
import java.util.concurrent.Future;
-import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
@@ -30,13 +29,13 @@
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSession.class);
private final KafkaProperties properties;
- private final Provider<KafkaProducer<String, String>> producerProvider;
+ private final Provider<Producer<String, String>> producerProvider;
private final KafkaEventsPublisherMetrics publisherMetrics;
private volatile Producer<String, String> producer;
@Inject
public KafkaSession(
- Provider<KafkaProducer<String, String>> producerProvider,
+ Provider<Producer<String, String>> producerProvider,
KafkaProperties properties,
KafkaEventsPublisherMetrics publisherMetrics) {
this.producerProvider = producerProvider;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventNativeSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventNativeSubscriber.java
new file mode 100644
index 0000000..a98e098
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventNativeSubscriber.java
@@ -0,0 +1,207 @@
+// Copyright (C) 2019 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.googlesource.gerrit.plugins.kafka.subscribe;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.server.util.ManualRequestContext;
+import com.google.gerrit.server.util.OneOffRequestContext;
+import com.google.inject.Inject;
+import com.googlesource.gerrit.plugins.kafka.broker.ConsumerExecutor;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.Deserializer;
+
+public class KafkaEventNativeSubscriber implements KafkaEventSubscriber {
+ private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+ private static final int DELAY_RECONNECT_AFTER_FAILURE_MSEC = 1000;
+
+ private final OneOffRequestContext oneOffCtx;
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ private final Deserializer<EventMessage> valueDeserializer;
+ private final KafkaSubscriberProperties configuration;
+ private final ExecutorService executor;
+ private final KafkaEventSubscriberMetrics subscriberMetrics;
+ private final KafkaConsumerFactory consumerFactory;
+ private final Deserializer<byte[]> keyDeserializer;
+
+ private java.util.function.Consumer<EventMessage> messageProcessor;
+ private String topic;
+ private AtomicBoolean resetOffset = new AtomicBoolean(false);
+
+ private volatile ReceiverJob receiver;
+
+ @Inject
+ public KafkaEventNativeSubscriber(
+ KafkaSubscriberProperties configuration,
+ KafkaConsumerFactory consumerFactory,
+ Deserializer<byte[]> keyDeserializer,
+ Deserializer<EventMessage> valueDeserializer,
+ OneOffRequestContext oneOffCtx,
+ @ConsumerExecutor ExecutorService executor,
+ KafkaEventSubscriberMetrics subscriberMetrics) {
+
+ this.configuration = configuration;
+ this.oneOffCtx = oneOffCtx;
+ this.executor = executor;
+ this.subscriberMetrics = subscriberMetrics;
+ this.consumerFactory = consumerFactory;
+ this.keyDeserializer = keyDeserializer;
+ this.valueDeserializer = valueDeserializer;
+ }
+
+ /* (non-Javadoc)
+ * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#subscribe(java.lang.String, java.util.function.Consumer)
+ */
+ @Override
+ public void subscribe(String topic, java.util.function.Consumer<EventMessage> messageProcessor) {
+ this.topic = topic;
+ this.messageProcessor = messageProcessor;
+ logger.atInfo().log(
+ "Kafka consumer subscribing to topic alias [%s] for event topic [%s]", topic, topic);
+ runReceiver();
+ }
+
+ private void runReceiver() {
+ final ClassLoader previousClassLoader = Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread()
+ .setContextClassLoader(KafkaEventNativeSubscriber.class.getClassLoader());
+ Consumer<byte[], byte[]> consumer = consumerFactory.create(keyDeserializer);
+ consumer.subscribe(Collections.singleton(topic));
+ receiver = new ReceiverJob(consumer);
+ executor.execute(receiver);
+ } finally {
+ Thread.currentThread().setContextClassLoader(previousClassLoader);
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#shutdown()
+ */
+ @Override
+ public void shutdown() {
+ closed.set(true);
+ receiver.wakeup();
+ }
+
+ /* (non-Javadoc)
+ * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#getMessageProcessor()
+ */
+ @Override
+ public java.util.function.Consumer<EventMessage> getMessageProcessor() {
+ return messageProcessor;
+ }
+
+ /* (non-Javadoc)
+ * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#getTopic()
+ */
+ @Override
+ public String getTopic() {
+ return topic;
+ }
+
+ /* (non-Javadoc)
+ * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#resetOffset()
+ */
+ @Override
+ public void resetOffset() {
+ resetOffset.set(true);
+ }
+
+ private class ReceiverJob implements Runnable {
+ private final Consumer<byte[], byte[]> consumer;
+
+ public ReceiverJob(Consumer<byte[], byte[]> consumer) {
+ this.consumer = consumer;
+ }
+
+ public void wakeup() {
+ consumer.wakeup();
+ }
+
+ @Override
+ public void run() {
+ try {
+ consume();
+ } catch (Exception e) {
+ logger.atSevere().withCause(e).log("Consumer loop of topic %s ended", topic);
+ }
+ }
+
+ private void consume() throws InterruptedException {
+ try {
+ while (!closed.get()) {
+ if (resetOffset.getAndSet(false)) {
+ // Make sure there is an assignment for this consumer
+ while (consumer.assignment().isEmpty() && !closed.get()) {
+ logger.atInfo().log(
+ "Resetting offset: no partitions assigned to the consumer, request assignment.");
+ consumer.poll(Duration.ofMillis(configuration.getPollingInterval()));
+ }
+ consumer.seekToBeginning(consumer.assignment());
+ }
+ ConsumerRecords<byte[], byte[]> consumerRecords =
+ consumer.poll(Duration.ofMillis(configuration.getPollingInterval()));
+ consumerRecords.forEach(
+ consumerRecord -> {
+ try (ManualRequestContext ctx = oneOffCtx.open()) {
+ EventMessage event =
+ valueDeserializer.deserialize(consumerRecord.topic(), consumerRecord.value());
+ messageProcessor.accept(event);
+ } catch (Exception e) {
+ logger.atSevere().withCause(e).log(
+ "Malformed event '%s': [Exception: %s]",
+ new String(consumerRecord.value(), UTF_8));
+ subscriberMetrics.incrementSubscriberFailedToConsumeMessage();
+ }
+ });
+ }
+ } catch (WakeupException e) {
+ // Ignore exception if closing
+ if (!closed.get()) {
+ logger.atSevere().withCause(e).log("Consumer loop of topic %s interrupted", topic);
+ reconnectAfterFailure();
+ }
+ } catch (Exception e) {
+ subscriberMetrics.incrementSubscriberFailedToPollMessages();
+ logger.atSevere().withCause(e).log(
+ "Existing consumer loop of topic %s because of a non-recoverable exception", topic);
+ reconnectAfterFailure();
+ } finally {
+ consumer.close();
+ }
+ }
+
+ private void reconnectAfterFailure() throws InterruptedException {
+ // Random delay with average of DELAY_RECONNECT_AFTER_FAILURE_MSEC
+ // for avoiding hammering exactly at the same interval in case of failure
+ long reconnectDelay =
+ DELAY_RECONNECT_AFTER_FAILURE_MSEC / 2
+ + new Random().nextInt(DELAY_RECONNECT_AFTER_FAILURE_MSEC);
+ Thread.sleep(reconnectDelay);
+ runReceiver();
+ }
+ }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriber.java
index 7ef9d7b..6315dea 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriber.java
@@ -1,4 +1,4 @@
-// Copyright (C) 2019 The Android Open Source Project
+// Copyright (C) 2021 The Android Open Source Project
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
@@ -11,176 +11,39 @@
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
+
package com.googlesource.gerrit.plugins.kafka.subscribe;
-import static java.nio.charset.StandardCharsets.UTF_8;
-
import com.gerritforge.gerrit.eventbroker.EventMessage;
-import com.google.common.flogger.FluentLogger;
-import com.google.gerrit.server.util.ManualRequestContext;
-import com.google.gerrit.server.util.OneOffRequestContext;
-import com.google.inject.Inject;
-import com.googlesource.gerrit.plugins.kafka.broker.ConsumerExecutor;
-import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
-import java.time.Duration;
-import java.util.Collections;
-import java.util.Random;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.common.errors.WakeupException;
-import org.apache.kafka.common.serialization.Deserializer;
-public class KafkaEventSubscriber {
- private static final FluentLogger logger = FluentLogger.forEnclosingClass();
- private static final int DELAY_RECONNECT_AFTER_FAILURE_MSEC = 1000;
+/** Generic interface to a Kafka topic subscriber. */
+public interface KafkaEventSubscriber {
- private final OneOffRequestContext oneOffCtx;
- private final AtomicBoolean closed = new AtomicBoolean(false);
+ /**
+ * Subscribe to a topic and receive messages asynchronously.
+ *
+ * @param topic Kafka topic name
+ * @param messageProcessor consumer function for processing incoming messages
+ */
+ void subscribe(String topic, java.util.function.Consumer<EventMessage> messageProcessor);
- private final Deserializer<EventMessage> valueDeserializer;
- private final KafkaSubscriberProperties configuration;
- private final ExecutorService executor;
- private final KafkaEventSubscriberMetrics subscriberMetrics;
- private final KafkaConsumerFactory consumerFactory;
- private final Deserializer<byte[]> keyDeserializer;
+ /** Shutdown Kafka consumer. */
+ void shutdown();
- private java.util.function.Consumer<EventMessage> messageProcessor;
- private String topic;
- private AtomicBoolean resetOffset = new AtomicBoolean(false);
+ /**
+ * Returns the current consumer function for the subscribed topic.
+ *
+ * @return the default topic consumer function.
+ */
+ java.util.function.Consumer<EventMessage> getMessageProcessor();
- private volatile ReceiverJob receiver;
+ /**
+ * Returns the current subscribed topic name.
+ *
+ * @return Kafka topic name.
+ */
+ String getTopic();
- @Inject
- public KafkaEventSubscriber(
- KafkaSubscriberProperties configuration,
- KafkaConsumerFactory consumerFactory,
- Deserializer<byte[]> keyDeserializer,
- Deserializer<EventMessage> valueDeserializer,
- OneOffRequestContext oneOffCtx,
- @ConsumerExecutor ExecutorService executor,
- KafkaEventSubscriberMetrics subscriberMetrics) {
-
- this.configuration = configuration;
- this.oneOffCtx = oneOffCtx;
- this.executor = executor;
- this.subscriberMetrics = subscriberMetrics;
- this.consumerFactory = consumerFactory;
- this.keyDeserializer = keyDeserializer;
- this.valueDeserializer = valueDeserializer;
- }
-
- public void subscribe(String topic, java.util.function.Consumer<EventMessage> messageProcessor) {
- this.topic = topic;
- this.messageProcessor = messageProcessor;
- logger.atInfo().log(
- "Kafka consumer subscribing to topic alias [%s] for event topic [%s]", topic, topic);
- runReceiver();
- }
-
- private void runReceiver() {
- final ClassLoader previousClassLoader = Thread.currentThread().getContextClassLoader();
- try {
- Thread.currentThread().setContextClassLoader(KafkaEventSubscriber.class.getClassLoader());
- Consumer<byte[], byte[]> consumer = consumerFactory.create(keyDeserializer);
- consumer.subscribe(Collections.singleton(topic));
- receiver = new ReceiverJob(consumer);
- executor.execute(receiver);
- } finally {
- Thread.currentThread().setContextClassLoader(previousClassLoader);
- }
- }
-
- public void shutdown() {
- closed.set(true);
- receiver.wakeup();
- }
-
- public java.util.function.Consumer<EventMessage> getMessageProcessor() {
- return messageProcessor;
- }
-
- public String getTopic() {
- return topic;
- }
-
- public void resetOffset() {
- resetOffset.set(true);
- }
-
- private class ReceiverJob implements Runnable {
- private final Consumer<byte[], byte[]> consumer;
-
- public ReceiverJob(Consumer<byte[], byte[]> consumer) {
- this.consumer = consumer;
- }
-
- public void wakeup() {
- consumer.wakeup();
- }
-
- @Override
- public void run() {
- try {
- consume();
- } catch (Exception e) {
- logger.atSevere().withCause(e).log("Consumer loop of topic %s ended", topic);
- }
- }
-
- private void consume() throws InterruptedException {
- try {
- while (!closed.get()) {
- if (resetOffset.getAndSet(false)) {
- // Make sure there is an assignment for this consumer
- while (consumer.assignment().isEmpty() && !closed.get()) {
- logger.atInfo().log(
- "Resetting offset: no partitions assigned to the consumer, request assignment.");
- consumer.poll(Duration.ofMillis(configuration.getPollingInterval()));
- }
- consumer.seekToBeginning(consumer.assignment());
- }
- ConsumerRecords<byte[], byte[]> consumerRecords =
- consumer.poll(Duration.ofMillis(configuration.getPollingInterval()));
- consumerRecords.forEach(
- consumerRecord -> {
- try (ManualRequestContext ctx = oneOffCtx.open()) {
- EventMessage event =
- valueDeserializer.deserialize(consumerRecord.topic(), consumerRecord.value());
- messageProcessor.accept(event);
- } catch (Exception e) {
- logger.atSevere().withCause(e).log(
- "Malformed event '%s': [Exception: %s]",
- new String(consumerRecord.value(), UTF_8));
- subscriberMetrics.incrementSubscriberFailedToConsumeMessage();
- }
- });
- }
- } catch (WakeupException e) {
- // Ignore exception if closing
- if (!closed.get()) {
- logger.atSevere().withCause(e).log("Consumer loop of topic %s interrupted", topic);
- reconnectAfterFailure();
- }
- } catch (Exception e) {
- subscriberMetrics.incrementSubscriberFailedToPollMessages();
- logger.atSevere().withCause(e).log(
- "Existing consumer loop of topic %s because of a non-recoverable exception", topic);
- reconnectAfterFailure();
- } finally {
- consumer.close();
- }
- }
-
- private void reconnectAfterFailure() throws InterruptedException {
- // Random delay with average of DELAY_RECONNECT_AFTER_FAILURE_MSEC
- // for avoiding hammering exactly at the same interval in case of failure
- long reconnectDelay =
- DELAY_RECONNECT_AFTER_FAILURE_MSEC / 2
- + new Random().nextInt(DELAY_RECONNECT_AFTER_FAILURE_MSEC);
- Thread.sleep(reconnectDelay);
- runReceiver();
- }
- }
+ /** Reset the offset for reading incoming Kafka messages of the topic. */
+ void resetOffset();
}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java
index 5a85b81..af0d53f 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java
@@ -45,7 +45,7 @@
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
-import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.junit.After;
import org.junit.AfterClass;
@@ -103,8 +103,7 @@
new KafkaSubscriberProperties(
TEST_POLLING_INTERVAL_MSEC, TEST_GROUP_ID, TEST_NUM_SUBSCRIBERS);
bind(KafkaSubscriberProperties.class).toInstance(kafkaSubscriberProperties);
- bind(new TypeLiteral<KafkaProducer<String, String>>() {})
- .toProvider(KafkaProducerProvider.class);
+ bind(new TypeLiteral<Producer<String, String>>() {}).toProvider(KafkaProducerProvider.class);
bind(WorkQueue.class).to(TestWorkQueue.class);
}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaSessionTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaSessionTest.java
index 46eaa73..ccf60ce 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaSessionTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaSessionTest.java
@@ -25,7 +25,7 @@
import com.googlesource.gerrit.plugins.kafka.session.KafkaProducerProvider;
import com.googlesource.gerrit.plugins.kafka.session.KafkaSession;
import org.apache.kafka.clients.producer.Callback;
-import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.TopicPartition;
import org.junit.Before;
@@ -39,7 +39,7 @@
@RunWith(MockitoJUnitRunner.class)
public class KafkaSessionTest {
KafkaSession objectUnderTest;
- @Mock KafkaProducer<String, String> kafkaProducer;
+ @Mock Producer<String, String> kafkaProducer;
@Mock KafkaProducerProvider producerProvider;
@Mock KafkaProperties properties;
@Mock KafkaEventsPublisherMetrics publisherMetrics;
