Receive messages through Kafka REST API Allow to receive messages through the Confluent REST Proxy instead of using the Kafka native client. Also implemment the reset of the topic partitions offset for the replay of all messages in a topic. Bug: Issue 15164 Change-Id: I30142b88d8a215dc61261877cb8efe5d246320ff
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 d8e6f82..55ce631 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java
@@ -18,31 +18,41 @@ import com.google.gerrit.extensions.events.LifecycleListener; import com.google.gerrit.extensions.registration.DynamicSet; import com.google.gerrit.server.events.EventListener; +import com.google.gerrit.server.git.WorkQueue; import com.google.gson.Gson; import com.google.inject.AbstractModule; import com.google.inject.Inject; import com.google.inject.Scopes; import com.google.inject.Singleton; import com.google.inject.TypeLiteral; +import com.google.inject.assistedinject.FactoryModuleBuilder; import com.googlesource.gerrit.plugins.kafka.api.KafkaApiModule; import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties; import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties.ClientType; import com.googlesource.gerrit.plugins.kafka.config.RequestConfigProvider; import com.googlesource.gerrit.plugins.kafka.publish.KafkaPublisher; import com.googlesource.gerrit.plugins.kafka.publish.KafkaRestProducer; +import com.googlesource.gerrit.plugins.kafka.rest.FutureExecutor; +import com.googlesource.gerrit.plugins.kafka.rest.HttpHostProxy; +import com.googlesource.gerrit.plugins.kafka.rest.HttpHostProxyProvider; +import com.googlesource.gerrit.plugins.kafka.rest.KafkaRestClient; import com.googlesource.gerrit.plugins.kafka.session.KafkaProducerProvider; +import java.util.concurrent.ExecutorService; import org.apache.http.client.config.RequestConfig; import org.apache.kafka.clients.producer.Producer; class Module extends AbstractModule { + private static final int HTTP_THREAD_POOL_SIZE = 10; private final KafkaApiModule kafkaBrokerModule; private final KafkaProperties kafkaConf; + private final WorkQueue workQueue; @Inject - public Module(KafkaApiModule kafkaBrokerModule, KafkaProperties kafkaConf) { + public Module(KafkaApiModule kafkaBrokerModule, KafkaProperties kafkaConf, WorkQueue workQueue) { this.kafkaBrokerModule = kafkaBrokerModule; this.kafkaConf = kafkaConf; + this.workQueue = workQueue; } @Override @@ -58,8 +68,14 @@ .toProvider(KafkaProducerProvider.class); break; case REST: + bind(ExecutorService.class) + .annotatedWith(FutureExecutor.class) + .toInstance( + workQueue.createQueue(HTTP_THREAD_POOL_SIZE, "KafkaRestClientThreadPool", true)); + bind(HttpHostProxy.class).toProvider(HttpHostProxyProvider.class).in(Scopes.SINGLETON); bind(RequestConfig.class).toProvider(RequestConfigProvider.class).in(Scopes.SINGLETON); bind(new TypeLiteral<Producer<String, String>>() {}).to(KafkaRestProducer.class); + install(new FactoryModuleBuilder().build(KafkaRestClient.Factory.class)); break; default: throw new IllegalArgumentException("Unsupported Kafka client type " + clientType);
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 b76ff7f..47f9969 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
@@ -26,9 +26,11 @@ import com.google.inject.Singleton; import com.google.inject.TypeLiteral; import com.googlesource.gerrit.plugins.kafka.broker.ConsumerExecutor; +import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties.ClientType; 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.KafkaEventRestSubscriber; import com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber; import java.util.Set; import java.util.concurrent.ExecutorService; @@ -56,7 +58,17 @@ @Override protected void configure() { - bind(KafkaEventSubscriber.class).to(KafkaEventNativeSubscriber.class); + ClientType clientType = configuration.getClientType(); + switch (clientType) { + case NATIVE: + bind(KafkaEventSubscriber.class).to(KafkaEventNativeSubscriber.class); + break; + case REST: + bind(KafkaEventSubscriber.class).to(KafkaEventRestSubscriber.class); + break; + default: + throw new IllegalArgumentException("Unsupported Kafka client type " + clientType); + } bind(ExecutorService.class) .annotatedWith(ConsumerExecutor.class)
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java index 4c9f285..6a8137d 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java
@@ -34,6 +34,7 @@ public class KafkaProperties extends java.util.Properties { private static final long serialVersionUID = 0L; + private static final boolean DEFAULT_HTTP_WIRE_LOG = false; public static final String KAFKA_STRING_SERIALIZER = StringSerializer.class.getName(); public enum ClientType { @@ -45,6 +46,7 @@ private final boolean sendAsync; private final ClientType clientType; private final URI restApiUri; + private final boolean httpWireLog; @Inject public KafkaProperties(PluginConfigFactory configFactory, @PluginName String pluginName) { @@ -67,10 +69,12 @@ } catch (URISyntaxException e) { throw new IllegalArgumentException("Invalid Kafka REST API URI: " + restApiUriString, e); } + httpWireLog = fromGerritConfig.getBoolean("httpWireLog", DEFAULT_HTTP_WIRE_LOG); break; case NATIVE: default: restApiUri = null; + httpWireLog = false; break; } @@ -87,6 +91,7 @@ this.clientType = clientType; this.restApiUri = restApiURI; initDockerizedKafkaServer(); + this.httpWireLog = false; } private void setDefaults() { @@ -140,4 +145,8 @@ public URI getRestApiUri() { return restApiUri; } + + public boolean isHttpWireLog() { + return httpWireLog; + } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/RequestConfigProvider.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/RequestConfigProvider.java index 41e324c..361d01a 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/RequestConfigProvider.java +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/RequestConfigProvider.java
@@ -14,29 +14,24 @@ package com.googlesource.gerrit.plugins.kafka.config; -import com.google.gerrit.httpd.ProxyProperties; import com.google.inject.Inject; import com.google.inject.Provider; -import java.net.URL; -import java.util.Optional; -import org.apache.http.HttpHost; +import com.googlesource.gerrit.plugins.kafka.rest.HttpHostProxy; import org.apache.http.client.config.RequestConfig; import org.apache.http.client.config.RequestConfig.Builder; public class RequestConfigProvider implements Provider<RequestConfig> { - private final Optional<HttpHost> proxyHost; + private final HttpHostProxy proxyHost; @Inject - public RequestConfigProvider(ProxyProperties proxyConf) { - proxyHost = - Optional.ofNullable(proxyConf.getProxyUrl()).map(URL::toString).map(HttpHost::create); + public RequestConfigProvider(HttpHostProxy proxyHost) { + this.proxyHost = proxyHost; } @Override public RequestConfig get() { Builder configBuilder = RequestConfig.custom(); - configBuilder = proxyHost.map(configBuilder::setProxy).orElse(configBuilder); - return configBuilder.build(); + return proxyHost.apply(configBuilder).build(); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaRestProducer.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaRestProducer.java index 54e35fa..8799cef 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaRestProducer.java +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaRestProducer.java
@@ -16,29 +16,19 @@ import com.google.common.flogger.FluentLogger; import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.JdkFutureAdapters; -import com.google.common.util.concurrent.ListenableFuture; import com.google.inject.Inject; import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties; +import com.googlesource.gerrit.plugins.kafka.rest.KafkaRestClient; import java.io.IOException; -import java.net.URI; -import java.net.URLEncoder; import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; -import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import org.apache.http.HttpEntity; -import org.apache.http.HttpHeaders; -import org.apache.http.HttpResponse; import org.apache.http.HttpStatus; -import org.apache.http.client.config.RequestConfig; import org.apache.http.client.methods.HttpPost; import org.apache.http.entity.ContentType; import org.apache.http.entity.StringEntity; -import org.apache.http.impl.nio.client.CloseableHttpAsyncClient; -import org.apache.http.impl.nio.client.HttpAsyncClients; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.Producer; @@ -55,20 +45,11 @@ new RecordMetadata(null, 0, 0, 0, null, 0, 0); private static final FluentLogger logger = FluentLogger.forEnclosingClass(); private static final String KAFKA_V2_JSON = "application/vnd.kafka.json.v2+json"; - private final URI kafkaRestApi; - private final ExecutorService futureExecutor; - private final CloseableHttpAsyncClient httpclient; - private RequestConfig httpRequestConf; + private final KafkaRestClient restClient; @Inject - public KafkaRestProducer( - KafkaProperties kafkaConf, - RequestConfig httpRequestConf, - @FutureExecutor ExecutorService futureExecutor) { - this.kafkaRestApi = kafkaConf.getRestApiUri(); - this.futureExecutor = futureExecutor; - httpclient = HttpAsyncClients.createDefault(); - this.httpRequestConf = httpRequestConf; + public KafkaRestProducer(KafkaProperties kafkaConf, KafkaRestClient.Factory restClientFactory) { + restClient = restClientFactory.create(kafkaConf); } @Override @@ -105,27 +86,15 @@ @Override public Future<RecordMetadata> send(ProducerRecord<String, String> record, Callback callback) { - httpclient.start(); HttpPost post = - createPostToTopic( + restClient.createPostToTopic( record.topic(), new StringEntity( getRecordAsJson(record), ContentType.create(KAFKA_V2_JSON, StandardCharsets.UTF_8))); - return Futures.transformAsync( - JdkFutureAdapters.listenInPoolThread(httpclient.execute(post, null), futureExecutor), - this::getRecordMetadataResult, - futureExecutor); - } - - private HttpPost createPostToTopic(String topic, HttpEntity postBodyEntity) { - HttpPost post = - new HttpPost( - kafkaRestApi.resolve("/topics/" + URLEncoder.encode(topic, StandardCharsets.UTF_8))); - post.addHeader(HttpHeaders.ACCEPT, "*/*"); - post.setConfig(httpRequestConf); - post.setEntity(postBodyEntity); - return post; + return restClient.mapAsync( + restClient.execute(post, HttpStatus.SC_OK), + (res) -> Futures.immediateFuture(ZEROS_RECORD_METADATA)); } @Override @@ -146,7 +115,7 @@ @Override public void close() { try { - httpclient.close(); + restClient.close(); } catch (IOException e) { logger.atWarning().withCause(e).log("Unable to close httpclient"); } @@ -157,21 +126,6 @@ close(); } - private ListenableFuture<RecordMetadata> getRecordMetadataResult(HttpResponse response) { - switch (response.getStatusLine().getStatusCode()) { - case HttpStatus.SC_OK: - return Futures.immediateFuture(ZEROS_RECORD_METADATA); - default: - return Futures.immediateFailedFuture( - new IOException( - String.format( - "Request failed: HTTP status %d (%s)\n%s", - response.getStatusLine().getStatusCode(), - response.getStatusLine().getReasonPhrase(), - response.getEntity()))); - } - } - private String getRecordAsJson(ProducerRecord<String, String> record) { return String.format( "{\"records\":[{\"key\":\"%s\",\"value\":%s}]}", record.key(), record.value());
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/FutureExecutor.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/FutureExecutor.java similarity index 93% rename from src/main/java/com/googlesource/gerrit/plugins/kafka/publish/FutureExecutor.java rename to src/main/java/com/googlesource/gerrit/plugins/kafka/rest/FutureExecutor.java index af181da..2ff0840 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/FutureExecutor.java +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/FutureExecutor.java
@@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package com.googlesource.gerrit.plugins.kafka.publish; +package com.googlesource.gerrit.plugins.kafka.rest; import com.google.inject.BindingAnnotation; import java.lang.annotation.Retention;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpHostProxy.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpHostProxy.java new file mode 100644 index 0000000..595b95a --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpHostProxy.java
@@ -0,0 +1,56 @@ +// 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. +// 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.rest; + +import com.google.gerrit.common.Nullable; +import java.net.URL; +import org.apache.http.HttpHost; +import org.apache.http.auth.AuthScope; +import org.apache.http.auth.UsernamePasswordCredentials; +import org.apache.http.client.CredentialsProvider; +import org.apache.http.client.config.RequestConfig.Builder; +import org.apache.http.impl.client.BasicCredentialsProvider; +import org.apache.http.impl.nio.client.HttpAsyncClientBuilder; + +public class HttpHostProxy { + private final URL proxyUrl; + private final String username; + private final String password; + + public HttpHostProxy(URL proxyUrl, @Nullable String username, @Nullable String password) { + this.proxyUrl = proxyUrl; + this.username = username; + this.password = password; + } + + public Builder apply(Builder clientBuilder) { + if (proxyUrl != null) { + clientBuilder.setProxy( + new HttpHost(proxyUrl.getHost(), proxyUrl.getPort(), proxyUrl.getProtocol())); + } + return clientBuilder; + } + + public HttpAsyncClientBuilder apply(HttpAsyncClientBuilder custom) { + if (proxyUrl != null && username != null && password != null) { + CredentialsProvider credsProvider = new BasicCredentialsProvider(); + credsProvider.setCredentials( + new AuthScope(proxyUrl.getHost(), proxyUrl.getPort()), + new UsernamePasswordCredentials(username, password)); + custom.setDefaultCredentialsProvider(credsProvider); + } + return custom; + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpHostProxyProvider.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpHostProxyProvider.java new file mode 100644 index 0000000..7c82ecd --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpHostProxyProvider.java
@@ -0,0 +1,54 @@ +// 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. +// 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.rest; + +import com.google.common.base.Strings; +import com.google.gerrit.server.config.GerritServerConfig; +import com.google.inject.Inject; +import com.google.inject.Provider; +import java.net.MalformedURLException; +import java.net.URL; +import org.eclipse.jgit.lib.Config; + +public class HttpHostProxyProvider implements Provider<HttpHostProxy> { + private URL proxyUrl; + private String proxyUser; + private String proxyPassword; + + @Inject + HttpHostProxyProvider(@GerritServerConfig Config config) throws MalformedURLException { + String proxyUrlStr = config.getString("http", null, "proxy"); + if (!Strings.isNullOrEmpty(proxyUrlStr)) { + proxyUrl = new URL(proxyUrlStr); + proxyUser = config.getString("http", null, "proxyUsername"); + proxyPassword = config.getString("http", null, "proxyPassword"); + String userInfo = proxyUrl.getUserInfo(); + if (userInfo != null) { + int c = userInfo.indexOf(':'); + if (0 < c) { + proxyUser = userInfo.substring(0, c); + proxyPassword = userInfo.substring(c + 1); + } else { + proxyUser = userInfo; + } + } + } + } + + @Override + public HttpHostProxy get() { + return new HttpHostProxy(proxyUrl, proxyUser, proxyPassword); + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java new file mode 100644 index 0000000..b5faa85 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java
@@ -0,0 +1,248 @@ +// 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. +// 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.rest; + +import static java.nio.charset.StandardCharsets.UTF_8; + +import com.google.common.base.Function; +import com.google.common.flogger.FluentLogger; +import com.google.common.net.MediaType; +import com.google.common.util.concurrent.AsyncFunction; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.JdkFutureAdapters; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.gerrit.common.Nullable; +import com.google.inject.Inject; +import com.google.inject.assistedinject.Assisted; +import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.net.URI; +import java.net.URLEncoder; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.stream.Collectors; +import org.apache.http.HttpEntity; +import org.apache.http.HttpHeaders; +import org.apache.http.HttpResponse; +import org.apache.http.HttpStatus; +import org.apache.http.client.config.RequestConfig; +import org.apache.http.client.config.RequestConfig.Builder; +import org.apache.http.client.methods.HttpDelete; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.client.methods.HttpRequestBase; +import org.apache.http.entity.ContentType; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.nio.client.CloseableHttpAsyncClient; +import org.apache.http.impl.nio.client.HttpAsyncClients; +import org.apache.log4j.Level; +import org.apache.log4j.Logger; + +public class KafkaRestClient { + private static final FluentLogger logger = FluentLogger.forEnclosingClass(); + private static final String KAFKA_V2_JSON = "application/vnd.kafka.json.v2+json"; + private static final String KAFKA_V2 = "application/vnd.kafka.v2+json"; + + private final HttpHostProxy proxy; + private final CloseableHttpAsyncClient httpclient; + private final URI kafkaRestApiUri; + private final ExecutorService futureExecutor; + + public interface Factory { + KafkaRestClient create(KafkaProperties configuration); + } + + @Inject + public KafkaRestClient( + HttpHostProxy httpHostProxy, + @FutureExecutor ExecutorService executor, + @Assisted KafkaProperties configuration) { + proxy = httpHostProxy; + httpclient = proxy.apply(HttpAsyncClients.custom()).build(); + httpclient.start(); + kafkaRestApiUri = configuration.getRestApiUri(); + if (configuration.isHttpWireLog()) { + enableHttpWireLog(); + } + this.futureExecutor = executor; + } + + public static void enableHttpWireLog() { + Logger.getLogger("org.apache.http.wire").setLevel(Level.TRACE); + } + + public ListenableFuture<HttpResponse> execute(HttpRequestBase request, int... expectedStatuses) { + return Futures.transformAsync( + listenableFutureOf(httpclient.execute(request, null)), + (res) -> { + IOException exc = + getResponseException( + String.format("HTTP %s %s FAILED", request.getMethod(), request.getURI()), + res, + expectedStatuses); + if (exc == null) { + return Futures.immediateFuture(res); + } + return Futures.immediateFailedFuture(exc); + }, + futureExecutor); + } + + public <I, O> ListenableFuture<O> mapAsync( + ListenableFuture<I> inputFuture, AsyncFunction<? super I, ? extends O> mapFunction) { + return Futures.transformAsync(inputFuture, mapFunction, futureExecutor); + } + + public <I, O> ListenableFuture<O> map( + ListenableFuture<I> inputFuture, Function<? super I, ? extends O> mapFunction) { + return Futures.transform(inputFuture, mapFunction, futureExecutor); + } + + public HttpGet createGetTopic(String topic) { + HttpGet get = new HttpGet(kafkaRestApiUri.resolve("/topics/" + topic)); + get.addHeader(HttpHeaders.ACCEPT, KAFKA_V2_JSON); + get.setConfig(createRequestConfig()); + return get; + } + + public HttpGet createGetRecords(URI consumerUri) { + HttpGet get = new HttpGet(consumerUri.resolve(consumerUri.getPath() + "/records")); + get.addHeader(HttpHeaders.ACCEPT, KAFKA_V2_JSON); + get.setConfig(createRequestConfig()); + return get; + } + + public HttpPost createPostToConsumer(String consumerGroup) { + HttpPost post = + new HttpPost( + kafkaRestApiUri.resolve( + kafkaRestApiUri.getPath() + + "/consumers/" + + URLEncoder.encode(consumerGroup, UTF_8))); + post.addHeader(HttpHeaders.ACCEPT, MediaType.ANY_TYPE.toString()); + post.setConfig(createRequestConfig()); + post.setEntity( + new StringEntity( + "{\"format\": \"json\",\"auto.offset.reset\": \"earliest\"}", + ContentType.create(KAFKA_V2, UTF_8))); + return post; + } + + public HttpDelete createDeleteToConsumer(URI consumerUri) { + HttpDelete delete = new HttpDelete(consumerUri); + delete.addHeader(HttpHeaders.ACCEPT, "*/*"); + delete.setConfig(createRequestConfig()); + return delete; + } + + public HttpPost createPostToSubscribe(URI consumerUri, String topic) { + HttpPost post = new HttpPost(consumerUri.resolve(consumerUri.getPath() + "/subscription")); + post.addHeader(HttpHeaders.ACCEPT, "*/*"); + post.setConfig(createRequestConfig()); + post.setEntity( + new StringEntity( + String.format("{\"topics\":[\"%s\"]}", topic), ContentType.create(KAFKA_V2, UTF_8))); + return post; + } + + public HttpPost createPostToTopic(String topic, HttpEntity postBodyEntity) { + HttpPost post = + new HttpPost(kafkaRestApiUri.resolve("/topics/" + URLEncoder.encode(topic, UTF_8))); + post.addHeader(HttpHeaders.ACCEPT, "*/*"); + post.setConfig(createRequestConfig()); + post.setEntity(postBodyEntity); + return post; + } + + public HttpPost createPostSeekTopicFromBeginning( + URI consumerInstanceURI, String topic, Set<Integer> partitions) { + HttpPost post = new HttpPost(consumerInstanceURI.resolve("/positions/beginning")); + post.addHeader(HttpHeaders.ACCEPT, "*/*"); + post.setConfig(createRequestConfig()); + post.setEntity( + new StringEntity( + String.format( + "{\"partitions\",[%s]}", + partitions.stream() + .map( + partition -> + String.format("{\"topic\":\"%s\",\"partition\":%d}", topic, partition)) + .collect(Collectors.joining(","))), + UTF_8)); + return post; + } + + @Nullable + public IOException getResponseException( + String errorMessage, HttpResponse response, int... okHttpStatuses) { + int responseHttpStatus = response.getStatusLine().getStatusCode(); + if (okHttpStatuses.length == 0) { + okHttpStatuses = + new int[] {HttpStatus.SC_OK, HttpStatus.SC_CREATED, HttpStatus.SC_NO_CONTENT}; + } + for (int httpStatus : okHttpStatuses) { + if (responseHttpStatus == httpStatus) { + return null; + } + } + + String responseBody = ""; + try { + responseBody = getStringEntity(response); + } catch (IOException e) { + logger.atWarning().withCause(e).log( + "Unable to extrace the string entity for response %d (%s)", + response.getStatusLine().getStatusCode(), response.getStatusLine().getReasonPhrase()); + } + + return new IOException( + String.format( + "%s\nHTTP status %d (%s)\n%s", + errorMessage, + response.getStatusLine().getStatusCode(), + response.getStatusLine().getReasonPhrase(), + responseBody)); + } + + protected String getStringEntity(HttpResponse response) throws IOException { + HttpEntity entity = response.getEntity(); + try (ByteArrayOutputStream outStream = new ByteArrayOutputStream()) { + entity.writeTo(outStream); + outStream.close(); + return outStream.toString(UTF_8); + } + } + + private <V> ListenableFuture<V> listenableFutureOf(Future<V> future) { + return JdkFutureAdapters.listenInPoolThread(future, futureExecutor); + } + + private RequestConfig createRequestConfig() { + Builder configBuilder = RequestConfig.custom(); + configBuilder = proxy.apply(configBuilder); + RequestConfig config = configBuilder.build(); + return config; + } + + public void close() throws IOException { + httpclient.close(); + } + + public URI resolveURI(String path) { + return kafkaRestApiUri.resolve(path); + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventRestSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventRestSubscriber.java new file mode 100644 index 0000000..bab7cd3 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventRestSubscriber.java
@@ -0,0 +1,324 @@ +// 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. +// 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.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.gerrit.server.util.ManualRequestContext; +import com.google.gerrit.server.util.OneOffRequestContext; +import com.google.gson.Gson; +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import com.google.inject.Inject; +import com.googlesource.gerrit.plugins.kafka.broker.ConsumerExecutor; +import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties; +import com.googlesource.gerrit.plugins.kafka.rest.KafkaRestClient; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.Reader; +import java.net.URI; +import java.net.URISyntaxException; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import java.util.stream.StreamSupport; +import org.apache.http.HttpResponse; +import org.apache.http.HttpStatus; +import org.apache.http.client.methods.HttpDelete; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.methods.HttpPost; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.Deserializer; + +public class KafkaEventRestSubscriber 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 Gson gson; + + private java.util.function.Consumer<EventMessage> messageProcessor; + private String topic; + private final KafkaRestClient restClient; + private final AtomicBoolean resetOffset; + private volatile ReceiverJob receiver; + + @Inject + public KafkaEventRestSubscriber( + KafkaSubscriberProperties configuration, + Deserializer<EventMessage> valueDeserializer, + OneOffRequestContext oneOffCtx, + @ConsumerExecutor ExecutorService executor, + KafkaEventSubscriberMetrics subscriberMetrics, + KafkaRestClient.Factory restClientFactory) { + + this.configuration = configuration; + this.oneOffCtx = oneOffCtx; + this.executor = executor; + this.subscriberMetrics = subscriberMetrics; + this.valueDeserializer = valueDeserializer; + + gson = new Gson(); + restClient = restClientFactory.create(configuration); + resetOffset = new AtomicBoolean(false); + } + + /* (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); + try { + runReceiver(); + } catch (InterruptedException | ExecutionException e) { + throw new IllegalStateException(e); + } + } + + private void runReceiver() throws InterruptedException, ExecutionException { + receiver = new ReceiverJob(configuration.getGroupId()); + executor.execute(receiver); + } + + /* (non-Javadoc) + * @see com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber#shutdown() + */ + @Override + public void shutdown() { + try { + closed.set(true); + receiver.close(); + } catch (InterruptedException | ExecutionException | IOException e) { + logger.atWarning().withCause(e).log("Unable to close receiver for topic=%s", topic); + } + } + + /* (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 ListenableFuture<URI> kafkaRestConsumerUri; + private final ListenableFuture<?> kafkaSubscriber; + + public ReceiverJob(String consumerGroup) throws InterruptedException, ExecutionException { + kafkaRestConsumerUri = createConsumer(consumerGroup); + kafkaSubscriber = restClient.mapAsync(kafkaRestConsumerUri, this::subscribeToTopic); + kafkaSubscriber.get(); + } + + public void close() throws InterruptedException, ExecutionException, IOException { + restClient.mapAsync(kafkaRestConsumerUri, this::deleteConsumer).get(); + restClient.close(); + } + + @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, ExecutionException { + try { + while (!closed.get()) { + if (resetOffset.getAndSet(false)) { + restClient.map(getTopicPartitions(), this::seekToBeginning).get(); + } + + ConsumerRecords<byte[], byte[]> records = + restClient.mapAsync(kafkaRestConsumerUri, this::getRecords).get(); + records.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 (Exception e) { + subscriberMetrics.incrementSubscriberFailedToPollMessages(); + logger.atSevere().withCause(e).log( + "Existing consumer loop of topic %s because of a non-recoverable exception", topic); + reconnectAfterFailure(); + } finally { + restClient.mapAsync(kafkaRestConsumerUri, this::deleteConsumer).get(); + } + } + + private Void seekToBeginning(Set<Integer> partitions) { + ListenableFuture<HttpPost> post = + restClient.map( + kafkaRestConsumerUri, + uri -> restClient.createPostSeekTopicFromBeginning(uri, topic, partitions)); + restClient.map(post, restClient::execute); + return null; + } + + private ListenableFuture<Set<Integer>> getTopicPartitions() { + HttpGet getTopic = restClient.createGetTopic(topic); + return restClient.mapAsync( + restClient.execute(getTopic, HttpStatus.SC_OK), this::getPartitions); + } + + private ListenableFuture<ConsumerRecords<byte[], byte[]>> getRecords(URI consumerUri) { + HttpGet getRecords = restClient.createGetRecords(consumerUri); + return restClient.mapAsync( + restClient.execute(getRecords, HttpStatus.SC_OK), this::convertRecords); + } + + private ListenableFuture<HttpResponse> subscribeToTopic(URI consumerUri) { + HttpPost post = restClient.createPostToSubscribe(consumerUri, topic); + return restClient.execute(post); + } + + private ListenableFuture<?> deleteConsumer(URI consumerUri) { + HttpDelete delete = restClient.createDeleteToConsumer(consumerUri); + return restClient.execute(delete); + } + + private ListenableFuture<URI> createConsumer(String name) { + HttpPost post = restClient.createPostToConsumer(name); + return restClient.mapAsync(restClient.execute(post, HttpStatus.SC_OK), this::getConsumerUri); + } + + private ListenableFuture<Set<Integer>> getPartitions(HttpResponse response) { + try (Reader bodyReader = + new InputStreamReader(response.getEntity().getContent(), StandardCharsets.UTF_8)) { + JsonObject responseJson = gson.fromJson(bodyReader, JsonObject.class); + Set<Integer> partitions = extractPartitions(responseJson); + return Futures.immediateFuture(partitions); + } catch (IOException e) { + return Futures.immediateFailedFuture(e); + } + } + + private ListenableFuture<URI> getConsumerUri(HttpResponse response) { + try (Reader bodyReader = + new InputStreamReader(response.getEntity().getContent(), StandardCharsets.UTF_8)) { + JsonObject responseJson = gson.fromJson(bodyReader, JsonObject.class); + URI consumerUri = new URI(responseJson.get("base_uri").getAsString()); + return Futures.immediateFuture(restClient.resolveURI(consumerUri.getPath())); + } catch (UnsupportedOperationException | IOException | URISyntaxException e) { + return Futures.immediateFailedFuture(e); + } + } + + private ListenableFuture<ConsumerRecords<byte[], byte[]>> convertRecords( + HttpResponse response) { + try (Reader bodyReader = new InputStreamReader(response.getEntity().getContent())) { + JsonArray jsonRecords = gson.fromJson(bodyReader, JsonArray.class); + if (jsonRecords.size() == 0) { + return Futures.immediateFuture(new ConsumerRecords<>(Collections.emptyMap())); + } + + Stream<ConsumerRecord<byte[], byte[]>> jsonObjects = + StreamSupport.stream(jsonRecords.spliterator(), false) + .map(JsonElement::getAsJsonObject) + .map(this::jsonToConsumerRecords); + + Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> records = + jsonObjects.collect(Collectors.groupingBy(this::jsonRecordPartition)); + return Futures.immediateFuture(new ConsumerRecords<>(records)); + } catch (IOException e) { + subscriberMetrics.incrementSubscriberFailedToConsumeMessage(); + return Futures.immediateFailedFuture(e); + } + } + + private ConsumerRecord<byte[], byte[]> jsonToConsumerRecords(JsonObject jsonRecord) { + return new ConsumerRecord<>( + jsonRecord.get("topic").getAsString(), + jsonRecord.get("partition").getAsInt(), + jsonRecord.get("offset").getAsLong(), + jsonRecord.get("key").toString().getBytes(), + jsonRecord.get("value").toString().getBytes()); + } + + private Set<Integer> extractPartitions(JsonObject jsonRecord) { + return StreamSupport.stream( + jsonRecord.get("partitions").getAsJsonArray().spliterator(), false) + .map(jsonElem -> jsonElem.getAsJsonObject().get("partition")) + .map(JsonElement::getAsInt) + .collect(Collectors.toSet()); + } + + private TopicPartition jsonRecordPartition(ConsumerRecord<byte[], byte[]> consumerRecord) { + return new TopicPartition(topic, consumerRecord.partition()); + } + + private void reconnectAfterFailure() throws InterruptedException, ExecutionException { + // 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/resources/Documentation/config.md b/src/main/resources/Documentation/config.md index c8cf2ec..4afb236 100644 --- a/src/main/resources/Documentation/config.md +++ b/src/main/resources/Documentation/config.md
@@ -48,6 +48,12 @@ : Kafka consumer group for receiving messages. Default: Gerrit instance-id +`plugin.@PLUGIN@.httpWireLog` +: Enable the HTTP wire protocol logging in error_log for all the communication with + the [Confluent REST-API Proxy](https://docs.confluent.io/platform/current/kafka-rest/index.html). + **NOTE**: when `plugin.@PLUGIN@.restApiUri` is unset or set to `NATIVE`, this setting is ignored. + Default: false + `plugin.@PLUGIN@.pollingIntervalMs` : Polling interval in msec for receiving messages from Kafka topic subscription. Default: 1000 @@ -55,7 +61,7 @@ `plugin.@PLUGIN@.restApiUri` : URL of the [Confluent REST-API Proxy](https://docs.confluent.io/platform/current/kafka-rest/index.html) - for sending messages through REST-API instead of using the native Kafka client. + for sending/receiving messages through REST-API instead of using the native Kafka client. **NOTE**: when `plugin.@PLUGIN@.restApiUri` is unset or set to `NATIVE`, this setting is ignored. Default: unset
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 ee0f1f1..2a2d6d4 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
@@ -238,6 +238,28 @@ assertNoMoreExpectedMessages(testConsumer); } + @Test + public void shouldSendToTopicAndResetOffset() { + connectToKafka(new KafkaProperties(false, clientType, getKafkaRestApiURI())); + KafkaBrokerApi kafkaBrokerApi = injector.getInstance(KafkaBrokerApi.class); + String testTopic = "test_topic_reset"; + TestConsumer testConsumer = new TestConsumer(1); + EventMessage testEventMessage = new EventMessage(new TestHeader(), new ProjectCreatedEvent()); + + kafkaBrokerApi.receiveAsync(testTopic, testConsumer); + kafkaBrokerApi.send(testTopic, testEventMessage); + + assertThat(testConsumer.await()).isTrue(); + assertThat(testConsumer.messages).hasSize(1); + assertThat(gson.toJson(testConsumer.messages.get(0))).isEqualTo(gson.toJson(testEventMessage)); + + kafkaBrokerApi.replayAllEvents(testTopic); + + assertThat(testConsumer.await()).isTrue(); + assertThat(testConsumer.messages).hasSize(1); + assertThat(gson.toJson(testConsumer.messages.get(0))).isEqualTo(gson.toJson(testEventMessage)); + } + protected URI getKafkaRestApiURI() { return null; }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java index 0b57d05..d5f3c64 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java
@@ -14,15 +14,16 @@ package com.googlesource.gerrit.plugins.kafka.api; -import com.google.gerrit.httpd.ProxyProperties; import com.google.inject.TypeLiteral; +import com.google.inject.assistedinject.FactoryModuleBuilder; import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties; import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties.ClientType; import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties; -import com.googlesource.gerrit.plugins.kafka.publish.FutureExecutor; import com.googlesource.gerrit.plugins.kafka.publish.KafkaRestProducer; +import com.googlesource.gerrit.plugins.kafka.rest.FutureExecutor; +import com.googlesource.gerrit.plugins.kafka.rest.HttpHostProxy; +import com.googlesource.gerrit.plugins.kafka.rest.KafkaRestClient; import java.net.URI; -import java.net.URL; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import org.apache.kafka.clients.producer.Producer; @@ -59,29 +60,14 @@ kafkaRest.getApiURI()); bind(KafkaSubscriberProperties.class).toInstance(kafkaSubscriberProperties); - bind(ProxyProperties.class) - .toInstance( - new ProxyProperties() { + bind(HttpHostProxy.class).toInstance(new HttpHostProxy(null, null, null)); - @Override - public URL getProxyUrl() { - return null; - } - - @Override - public String getUsername() { - return null; - } - - @Override - public String getPassword() { - return null; - } - }); + install(new FactoryModuleBuilder().build(KafkaRestClient.Factory.class)); } }; } + @Override protected URI getKafkaRestApiURI() { return kafkaRest.getApiURI(); }