From ccbf1ac813fbb4d73d3a9f71c4a1b6ad0bbeb3a8 Mon Sep 17 00:00:00 2001 From: Lukasz Druminski Date: Tue, 14 Mar 2017 10:22:24 +0100 Subject: [PATCH 1/4] resolves #740 | Topic name is not escaped from underscore in ZookeeperCounterReporter --- .../zookeeper/ZookeeperCounterReporter.java | 2 +- .../zookeeper/ZookeeperCounterReporterTest.java | 16 ++++++++-------- .../static/partials/modal/editGroup.html | 3 ++- .../tech/hermes/integration/MetricsTest.java | 16 ++++++++-------- 4 files changed, 19 insertions(+), 18 deletions(-) diff --git a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/counter/zookeeper/ZookeeperCounterReporter.java b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/counter/zookeeper/ZookeeperCounterReporter.java index db9f7945d1..76dbcfadb1 100644 --- a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/counter/zookeeper/ZookeeperCounterReporter.java +++ b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/counter/zookeeper/ZookeeperCounterReporter.java @@ -93,7 +93,7 @@ private static TopicName escapedTopicName(String qualifiedTopicName) { TopicName topicName = fromQualifiedName(qualifiedTopicName); return new TopicName( escapeMetricsReplacementChar(topicName.getGroupName()), - escapeMetricsReplacementChar(topicName.getName()) + topicName.getName() ); } diff --git a/hermes-common/src/test/java/pl/allegro/tech/hermes/common/metric/counter/zookeeper/ZookeeperCounterReporterTest.java b/hermes-common/src/test/java/pl/allegro/tech/hermes/common/metric/counter/zookeeper/ZookeeperCounterReporterTest.java index 5821b227b0..e3488e59d7 100644 --- a/hermes-common/src/test/java/pl/allegro/tech/hermes/common/metric/counter/zookeeper/ZookeeperCounterReporterTest.java +++ b/hermes-common/src/test/java/pl/allegro/tech/hermes/common/metric/counter/zookeeper/ZookeeperCounterReporterTest.java @@ -37,23 +37,23 @@ public class ZookeeperCounterReporterTest { public static final SortedMap EMPTY_GAUGES = null; public static final String GROUP_NAME_UNDERSCORE = "pl_allegro_tech_skylab"; public static final String GROUP_NAME = "pl.allegro.tech.skylab"; - public static final String TOPIC_NAME = "topic1"; + public static final String TOPIC_NAME_UNDERSCORE = "topic_1"; public static final String SUBSCRIPTION_NAME_UNDERSCORE = "subscription_name"; public static final String SUBSCRIPTION_NAME = "subscription.name"; - public static final TopicName QUALIFIED_TOPIC_NAME = new TopicName(GROUP_NAME, TOPIC_NAME); + public static final TopicName QUALIFIED_TOPIC_NAME = new TopicName(GROUP_NAME, TOPIC_NAME_UNDERSCORE); public static final long COUNT = 100L; public static final String GRAPHITE_PREFIX = "tech.hermes"; private static PathsCompiler pathsCompiler = new PathsCompiler("localhost.domain"); - public static final String METRIC_NAME_FOR_PUBLISHED = pathsCompiler.compile(PUBLISHED, - pathContext().withGroup(GROUP_NAME_UNDERSCORE).withTopic(TOPIC_NAME).build()); + public static final String METRIC_NAME_FOR_PUBLISHED = pathsCompiler.compile(PUBLISHED, pathContext() + .withGroup(GROUP_NAME_UNDERSCORE).withTopic(TOPIC_NAME_UNDERSCORE).build()); - public static final String METRIC_NAME_FOR_DELIVERED = pathsCompiler.compile(DELIVERED, - pathContext().withGroup(GROUP_NAME_UNDERSCORE).withTopic(TOPIC_NAME).withSubscription(SUBSCRIPTION_NAME_UNDERSCORE).build()); + public static final String METRIC_NAME_FOR_DELIVERED = pathsCompiler.compile(DELIVERED, pathContext() + .withGroup(GROUP_NAME_UNDERSCORE).withTopic(TOPIC_NAME_UNDERSCORE).withSubscription(SUBSCRIPTION_NAME_UNDERSCORE).build()); - public static final String METRIC_NAME_FOR_DISCARDED = pathsCompiler.compile(DISCARDED, - pathContext().withGroup(GROUP_NAME_UNDERSCORE).withTopic(TOPIC_NAME).withSubscription(SUBSCRIPTION_NAME_UNDERSCORE).build()); + public static final String METRIC_NAME_FOR_DISCARDED = pathsCompiler.compile(DISCARDED, pathContext() + .withGroup(GROUP_NAME_UNDERSCORE).withTopic(TOPIC_NAME_UNDERSCORE).withSubscription(SUBSCRIPTION_NAME_UNDERSCORE).build()); @Mock private CounterStorage counterStorage; diff --git a/hermes-console/static/partials/modal/editGroup.html b/hermes-console/static/partials/modal/editGroup.html index 8927db9bb1..c2c084cf5b 100644 --- a/hermes-console/static/partials/modal/editGroup.html +++ b/hermes-console/static/partials/modal/editGroup.html @@ -14,7 +14,8 @@

Edit group:
- +
diff --git a/integration/src/integration/java/pl/allegro/tech/hermes/integration/MetricsTest.java b/integration/src/integration/java/pl/allegro/tech/hermes/integration/MetricsTest.java index 2cc6958e02..3d92c54320 100644 --- a/integration/src/integration/java/pl/allegro/tech/hermes/integration/MetricsTest.java +++ b/integration/src/integration/java/pl/allegro/tech/hermes/integration/MetricsTest.java @@ -43,18 +43,18 @@ public void initializeAlways() { @Test public void shouldIncreaseTopicMetricsAfterMessageHasBeenPublished() { // given - Topic topic = operations.buildTopic("topicMetricsGroup", "topic"); + Topic topic = operations.buildTopic("pl.group", "topic_metrics"); operations.createSubscription(topic, "subscription", HTTP_ENDPOINT_URL); - graphiteEndpoint.returnMetricForTopic("topicMetricsGroup", "topic", 10, 15); + graphiteEndpoint.returnMetricForTopic("pl_group", "topic_metrics", 10, 15); remoteService.expectMessages(TestMessage.simple().body()); - assertThat(publisher.publish("topicMetricsGroup.topic", TestMessage.simple().body()).getStatus()) + assertThat(publisher.publish("pl.group.topic_metrics", TestMessage.simple().body()).getStatus()) .isEqualTo(CREATED.getStatusCode()); remoteService.waitUntilReceived(); wait.until(() -> { // when - TopicMetrics metrics = management.topic().getMetrics("topicMetricsGroup.topic"); + TopicMetrics metrics = management.topic().getMetrics("pl.group.topic_metrics"); // then assertThat(metrics.getRate()).isEqualTo("10"); @@ -67,18 +67,18 @@ public void shouldIncreaseTopicMetricsAfterMessageHasBeenPublished() { @Test public void shouldIncreaseSubscriptionDeliveredMetricsAfterMessageDelivered() { // given - Topic topic = operations.buildTopic("subscriptionMetricsGroup", "topic"); + Topic topic = operations.buildTopic("pl.group", "topic"); operations.createSubscription(topic, "subscription", HTTP_ENDPOINT_URL); - graphiteEndpoint.returnMetricForSubscription("subscriptionMetricsGroup", "topic", "subscription", 15); + graphiteEndpoint.returnMetricForSubscription("pl_group", "topic", "subscription", 15); remoteService.expectMessages(TestMessage.simple().body()); - assertThat(publisher.publish("subscriptionMetricsGroup.topic", TestMessage.simple().body()).getStatus()) + assertThat(publisher.publish("pl.group.topic", TestMessage.simple().body()).getStatus()) .isEqualTo(CREATED.getStatusCode()); remoteService.waitUntilReceived(); wait.until(() -> { // when - SubscriptionMetrics metrics = management.subscription().getMetrics("subscriptionMetricsGroup.topic", "subscription"); + SubscriptionMetrics metrics = management.subscription().getMetrics("pl.group.topic", "subscription"); // then assertThat(metrics.getRate()).isEqualTo("15"); From e304e841ff217cd97f73ab07ddd61125d9b89466 Mon Sep 17 00:00:00 2001 From: Robert Date: Tue, 14 Mar 2017 10:33:56 +0100 Subject: [PATCH 2/4] payload content-type check & error handling message payload content-type check & error message for wrong content-type (#728) --- .../tech/hermes/api}/AvroMediaType.java | 2 +- .../hermes/api/endpoints/TopicEndpoint.java | 4 +-- .../UnsupportedContentTypeException.java | 8 ++++++ .../ApacheHttpClientMessageBatchSender.java | 2 +- .../sender/http/HttpRequestFactory.java | 2 +- .../message/MessageContentTypeEnforcer.java | 26 ++++++++++++++---- .../publishing/message/MessageFactory.java | 2 +- .../MessageContentTypeEnforcerTest.java | 27 ++++++++++++++++--- .../integration/PublishingAvroTest.java | 19 +++++++++++++ 9 files changed, 78 insertions(+), 14 deletions(-) rename {hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http => hermes-api/src/main/java/pl/allegro/tech/hermes/api}/AvroMediaType.java (59%) diff --git a/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/AvroMediaType.java b/hermes-api/src/main/java/pl/allegro/tech/hermes/api/AvroMediaType.java similarity index 59% rename from hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/AvroMediaType.java rename to hermes-api/src/main/java/pl/allegro/tech/hermes/api/AvroMediaType.java index edd9561de6..17f2458b50 100644 --- a/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/AvroMediaType.java +++ b/hermes-api/src/main/java/pl/allegro/tech/hermes/api/AvroMediaType.java @@ -1,4 +1,4 @@ -package pl.allegro.tech.hermes.consumers.consumer.sender.http; +package pl.allegro.tech.hermes.api; public class AvroMediaType { diff --git a/hermes-api/src/main/java/pl/allegro/tech/hermes/api/endpoints/TopicEndpoint.java b/hermes-api/src/main/java/pl/allegro/tech/hermes/api/endpoints/TopicEndpoint.java index 3bbe94671a..9c2811c155 100644 --- a/hermes-api/src/main/java/pl/allegro/tech/hermes/api/endpoints/TopicEndpoint.java +++ b/hermes-api/src/main/java/pl/allegro/tech/hermes/api/endpoints/TopicEndpoint.java @@ -14,11 +14,11 @@ import javax.ws.rs.PathParam; import javax.ws.rs.Produces; import javax.ws.rs.QueryParam; -import javax.ws.rs.core.MediaType; import javax.ws.rs.core.Response; import java.util.List; import static javax.ws.rs.core.MediaType.APPLICATION_JSON; +import static pl.allegro.tech.hermes.api.AvroMediaType.AVRO_BINARY; @Path("topics") public interface TopicEndpoint { @@ -69,7 +69,7 @@ List queryList( Response publishMessage(@PathParam("topicName") String qualifiedTopicName, String message); @POST - @Consumes(MediaType.TEXT_PLAIN) + @Consumes(AVRO_BINARY) @Produces(APPLICATION_JSON) @Path("/{topicName}") Response publishMessage(@PathParam("topicName") String qualifiedTopicName, byte[] message); diff --git a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/message/wrapper/UnsupportedContentTypeException.java b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/message/wrapper/UnsupportedContentTypeException.java index 81f6f8531c..29788ee711 100644 --- a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/message/wrapper/UnsupportedContentTypeException.java +++ b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/message/wrapper/UnsupportedContentTypeException.java @@ -22,4 +22,12 @@ public UnsupportedContentTypeException(Subscription subscription) { )); } + public UnsupportedContentTypeException(String payloadContentType, Topic topic) { + super(String.format( + "Unsupported payload content type header %s for topic %s", + payloadContentType, + topic.getQualifiedName() + )); + } + } diff --git a/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/ApacheHttpClientMessageBatchSender.java b/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/ApacheHttpClientMessageBatchSender.java index bef74232d2..e4156be9b8 100644 --- a/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/ApacheHttpClientMessageBatchSender.java +++ b/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/ApacheHttpClientMessageBatchSender.java @@ -20,7 +20,7 @@ import static pl.allegro.tech.hermes.api.ContentType.AVRO; import static pl.allegro.tech.hermes.common.http.MessageMetadataHeaders.BATCH_ID; import static pl.allegro.tech.hermes.common.http.MessageMetadataHeaders.RETRY_COUNT; -import static pl.allegro.tech.hermes.consumers.consumer.sender.http.AvroMediaType.AVRO_BINARY; +import static pl.allegro.tech.hermes.api.AvroMediaType.AVRO_BINARY; public class ApacheHttpClientMessageBatchSender implements MessageBatchSender { diff --git a/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/HttpRequestFactory.java b/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/HttpRequestFactory.java index 50039525a5..db153e4c66 100644 --- a/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/HttpRequestFactory.java +++ b/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/HttpRequestFactory.java @@ -21,7 +21,7 @@ import static pl.allegro.tech.hermes.common.http.MessageMetadataHeaders.MESSAGE_ID; import static pl.allegro.tech.hermes.common.http.MessageMetadataHeaders.RETRY_COUNT; import static pl.allegro.tech.hermes.common.http.MessageMetadataHeaders.SCHEMA_VERSION; -import static pl.allegro.tech.hermes.consumers.consumer.sender.http.AvroMediaType.AVRO_BINARY; +import static pl.allegro.tech.hermes.api.AvroMediaType.AVRO_BINARY; class HttpRequestFactory { diff --git a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageContentTypeEnforcer.java b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageContentTypeEnforcer.java index c8eabb11b5..abca33c582 100644 --- a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageContentTypeEnforcer.java +++ b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageContentTypeEnforcer.java @@ -1,25 +1,41 @@ package pl.allegro.tech.hermes.frontend.publishing.message; import org.apache.avro.Schema; +import org.apache.commons.lang.StringUtils; +import pl.allegro.tech.hermes.api.Topic; +import pl.allegro.tech.hermes.common.message.wrapper.UnsupportedContentTypeException; import tech.allegro.schema.json2avro.converter.JsonAvroConverter; import static javax.ws.rs.core.MediaType.APPLICATION_JSON; +import static pl.allegro.tech.hermes.api.AvroMediaType.AVRO_BINARY; public class MessageContentTypeEnforcer { private final JsonAvroConverter converter = new JsonAvroConverter(); private static final String APPLICATION_JSON_WITH_DELIM = APPLICATION_JSON + ";"; + private static final String AVRO_BINARY_WITH_DELIM = AVRO_BINARY + ";"; - public byte[] enforceAvro(String payloadContentType, byte[] data, Schema schema) { - if (isJSON(payloadContentType)) { + public byte[] enforceAvro(String payloadContentType, byte[] data, Schema schema, Topic topic) { + String contentTypeLowerCase = StringUtils.lowerCase(payloadContentType); + if (isJSON(contentTypeLowerCase)) { return converter.convertToAvro(data, schema); + } else if (isAvro(contentTypeLowerCase)) { + return data; + } else { + throw new UnsupportedContentTypeException(payloadContentType, topic); } - return data; } private boolean isJSON(String contentType) { - return contentType != null && (contentType.length() > APPLICATION_JSON.length() ? - contentType.startsWith(APPLICATION_JSON_WITH_DELIM) : contentType.equals(APPLICATION_JSON)); + return isOfType(contentType, APPLICATION_JSON, APPLICATION_JSON_WITH_DELIM); + } + + private boolean isAvro(String contentType) { + return isOfType(contentType, AVRO_BINARY, AVRO_BINARY_WITH_DELIM); + } + + private boolean isOfType(String contentType, String expectedContentType, String expectedWithDelim) { + return contentType != null && (contentType.equals(expectedContentType) || contentType.startsWith(expectedWithDelim)); } } diff --git a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageFactory.java b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageFactory.java index 98b9f7ddcb..061d6b39f0 100644 --- a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageFactory.java +++ b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageFactory.java @@ -101,7 +101,7 @@ private AvroMessage createAvroMessage(HeaderMap headerMap, Topic topic, String m AvroMessage message = new AvroMessage( messageId, - enforcer.enforceAvro(headerMap.getFirst(Headers.CONTENT_TYPE_STRING), messageContent, schema.getSchema()), + enforcer.enforceAvro(headerMap.getFirst(Headers.CONTENT_TYPE_STRING), messageContent, schema.getSchema(), topic), timestamp, schema); diff --git a/hermes-frontend/src/test/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageContentTypeEnforcerTest.java b/hermes-frontend/src/test/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageContentTypeEnforcerTest.java index 9637483ece..dba4809260 100644 --- a/hermes-frontend/src/test/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageContentTypeEnforcerTest.java +++ b/hermes-frontend/src/test/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageContentTypeEnforcerTest.java @@ -2,10 +2,15 @@ import org.apache.avro.Schema; import org.junit.Test; +import pl.allegro.tech.hermes.api.ContentType; +import pl.allegro.tech.hermes.api.Topic; +import pl.allegro.tech.hermes.common.message.wrapper.UnsupportedContentTypeException; import pl.allegro.tech.hermes.schema.CompiledSchema; import pl.allegro.tech.hermes.schema.SchemaVersion; import pl.allegro.tech.hermes.test.helper.avro.AvroUser; +import pl.allegro.tech.hermes.test.helper.builder.TopicBuilder; +import javax.ws.rs.core.MediaType; import java.io.IOException; import static org.assertj.core.api.Assertions.assertThat; @@ -14,13 +19,14 @@ public class MessageContentTypeEnforcerTest { private MessageContentTypeEnforcer enforcer = new MessageContentTypeEnforcer(); + private Topic topic = TopicBuilder.topic("test.Topic").withContentType(ContentType.AVRO).build(); private AvroUser avroMessage = new AvroUser("Bob", 30, "black"); private CompiledSchema schema = new CompiledSchema<>(avroMessage.getSchema(), SchemaVersion.valueOf(0)); @Test public void shouldConvertToAvroWhenReceivedJSONOnAvroTopic() throws IOException { // when - byte[] enforcedMessage = enforcer.enforceAvro("application/json", avroMessage.asJson().getBytes(), schema.getSchema()); + byte[] enforcedMessage = enforcer.enforceAvro("application/json", avroMessage.asJson().getBytes(), schema.getSchema(), topic); // then assertThat(enforcedMessage).isEqualTo(avroMessage.asBytes()); @@ -29,7 +35,7 @@ public void shouldConvertToAvroWhenReceivedJSONOnAvroTopic() throws IOException @Test public void shouldStringContentTypeOfAdditionalOptionsWhenInterpretingIt() throws IOException { // when - byte[] enforcedMessage = enforcer.enforceAvro("application/json;encoding=utf-8", avroMessage.asJson().getBytes(), schema.getSchema()); + byte[] enforcedMessage = enforcer.enforceAvro("application/json;encoding=utf-8", avroMessage.asJson().getBytes(), schema.getSchema(), topic); // then assertThat(enforcedMessage).isEqualTo(avroMessage.asBytes()); @@ -38,10 +44,25 @@ public void shouldStringContentTypeOfAdditionalOptionsWhenInterpretingIt() throw @Test public void shouldNotConvertWhenReceivingAvroOnAvroTopic() throws IOException { // when - byte[] enforcedMessage = enforcer.enforceAvro("avro/binary", avroMessage.asBytes(), schema.getSchema()); + byte[] enforcedMessage = enforcer.enforceAvro("avro/binary", avroMessage.asBytes(), schema.getSchema(), topic); // then assertThat(enforcedMessage).isEqualTo(avroMessage.asBytes()); } + @Test + public void shouldBeCaseInsensitiveForPayloadContentType() throws IOException { + // when + byte[] enforcedMessage = enforcer.enforceAvro("AVRO/Binary", avroMessage.asBytes(), schema.getSchema(), topic); + + // then + assertThat(enforcedMessage).isEqualTo(avroMessage.asBytes()); + } + + @Test(expected = UnsupportedContentTypeException.class) + public void shouldThrowUnsupportedContentTypeExceptionWhenReceivedWrongContentType() throws IOException { + // when + enforcer.enforceAvro(MediaType.TEXT_PLAIN, avroMessage.asBytes(), schema.getSchema(), topic); + } + } \ No newline at end of file diff --git a/integration/src/integration/java/pl/allegro/tech/hermes/integration/PublishingAvroTest.java b/integration/src/integration/java/pl/allegro/tech/hermes/integration/PublishingAvroTest.java index 4582f080ce..28c126faef 100644 --- a/integration/src/integration/java/pl/allegro/tech/hermes/integration/PublishingAvroTest.java +++ b/integration/src/integration/java/pl/allegro/tech/hermes/integration/PublishingAvroTest.java @@ -23,10 +23,12 @@ import javax.ws.rs.core.Response; import java.io.IOException; import java.time.Clock; +import java.util.Collections; import java.util.UUID; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeoutException; +import static javax.ws.rs.core.MediaType.TEXT_PLAIN; import static javax.ws.rs.core.Response.Status.BAD_REQUEST; import static javax.ws.rs.core.Response.Status.CREATED; import static javax.ws.rs.core.Response.Status.INTERNAL_SERVER_ERROR; @@ -73,6 +75,23 @@ public void shouldPublishAvroAndConsumeJsonMessage() throws InterruptedException remoteService.waitUntilReceived(json -> assertThatJson(json).isEqualTo(user.asJson())); } + @Test + public void shouldNotPublishAvroWhenMessageIsNotJsonOrAvro() throws InterruptedException, ExecutionException, TimeoutException, IOException { + // given + Topic topic = operations.buildTopic(topic("publishAvroConsumeAvro.topic") + .withContentType(AVRO) + .build() + ); + operations.saveSchema(topic, user.getSchemaAsString()); + operations.createSubscription(topic, "subscription", HTTP_ENDPOINT_URL, ContentType.AVRO); + + // when + Response response = publisher.publish("publishAvroConsumeAvro.topic", user.asJson(), Collections.singletonMap("Content-Type", TEXT_PLAIN)); + + // then + assertThat(response).hasStatus(BAD_REQUEST); + } + @Test public void shouldPublishAvroAndConsumeAvroMessage() throws InterruptedException, ExecutionException, TimeoutException, IOException { // given From a6deb2c09068f11b0612b874a940b4ea68c92bac Mon Sep 17 00:00:00 2001 From: Robert Date: Wed, 15 Mar 2017 13:58:22 +0100 Subject: [PATCH 3/4] 634 added latency metrics for schema registry (#739) 634 added latency metrics for schema registry --- .../hermes/common/metric/HermesMetrics.java | 5 + .../tech/hermes/common/metric/Timers.java | 5 + .../metric/timer}/StartedTimersPair.java | 2 +- .../common/schema/RawSchemaClientFactory.java | 16 +++- .../ReadMetricsTrackingRawSchemaClient.java | 72 +++++++++++++++ .../common/schema/SchemaRepositoryType.java | 13 ++- ...dMetricsTrackingRawSchemaClientTest.groovy | 91 +++++++++++++++++++ .../frontend/buffer/BackupMessagesLoader.java | 9 +- .../hermes/frontend/metric/CachedTopic.java | 1 + .../publishing/handlers/ExchangeMetrics.java | 2 +- .../handlers/MessageReadHandler.java | 2 +- .../handlers/PublishingHandler.java | 2 +- .../publishing/message/MessageFactory.java | 2 +- .../buffer/BackupMessagesLoaderTest.java | 2 +- .../tech/hermes/metrics/PathContext.java | 16 +++- .../tech/hermes/metrics/PathsCompiler.java | 2 + 16 files changed, 221 insertions(+), 21 deletions(-) rename {hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/metric => hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/timer}/StartedTimersPair.java (88%) create mode 100644 hermes-common/src/main/java/pl/allegro/tech/hermes/common/schema/ReadMetricsTrackingRawSchemaClient.java create mode 100644 hermes-common/src/test/groovy/pl/allegro/tech/hermes/common/schema/ReadMetricsTrackingRawSchemaClientTest.groovy diff --git a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/HermesMetrics.java b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/HermesMetrics.java index 89d0fad70f..4048d8a6c2 100644 --- a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/HermesMetrics.java +++ b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/HermesMetrics.java @@ -9,6 +9,7 @@ import pl.allegro.tech.hermes.api.Subscription; import pl.allegro.tech.hermes.api.TopicName; import pl.allegro.tech.hermes.common.metric.timer.ConsumerLatencyTimer; +import pl.allegro.tech.hermes.common.schema.SchemaRepositoryType; import pl.allegro.tech.hermes.metrics.PathContext; import pl.allegro.tech.hermes.metrics.PathsCompiler; @@ -177,6 +178,10 @@ private String metricRegistryName(String metricDisplayName) { return pathCompiler.compile(metricDisplayName); } + public Timer schemaTimer(String schemaMetric, SchemaRepositoryType schemaRepoType) { + return metricRegistry.timer(pathCompiler.compile(schemaMetric, pathContext().withSchemaRepoType(schemaRepoType.toString()).build())); + } + public Timer executorDurationTimer(String executorName) { return metricRegistry.timer(pathCompiler.compile(Timers.EXECUTOR_DURATION, pathContext().withExecutorName(executorName).build())); } diff --git a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/Timers.java b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/Timers.java index 32592d5acb..92f790df3b 100644 --- a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/Timers.java +++ b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/Timers.java @@ -6,6 +6,7 @@ import static pl.allegro.tech.hermes.metrics.PathsCompiler.OAUTH_PROVIDER_NAME; import static pl.allegro.tech.hermes.metrics.PathsCompiler.SUBSCRIPTION; import static pl.allegro.tech.hermes.metrics.PathsCompiler.TOPIC; +import static pl.allegro.tech.hermes.metrics.PathsCompiler.SCHEMA_REPO_TYPE; public class Timers { @@ -33,6 +34,10 @@ public class Timers { READ_LATENCY = "read-latency", + SCHEMA = "schema." + SCHEMA_REPO_TYPE, + GET_SCHEMA_LATENCY = SCHEMA + ".get-schema", + GET_SCHEMA_VERSIONS_LATENCY = SCHEMA + ".get-schema-versions", + CONSUMER_WORKLOAD_REBALANCE_DURATION = "consumers-workload." + KAFKA_CLUSTER + ".selective.rebalance-duration", OAUTH_PROVIDER_TOKEN_REQUEST_LATENCY = "oauth.provider." + OAUTH_PROVIDER_NAME + ".token-request-latency", diff --git a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/metric/StartedTimersPair.java b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/timer/StartedTimersPair.java similarity index 88% rename from hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/metric/StartedTimersPair.java rename to hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/timer/StartedTimersPair.java index d81d4edd3c..9ce0f61e71 100644 --- a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/metric/StartedTimersPair.java +++ b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/timer/StartedTimersPair.java @@ -1,4 +1,4 @@ -package pl.allegro.tech.hermes.frontend.metric; +package pl.allegro.tech.hermes.common.metric.timer; import com.codahale.metrics.Timer; diff --git a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/schema/RawSchemaClientFactory.java b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/schema/RawSchemaClientFactory.java index 15306707e5..1b2221ad39 100644 --- a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/schema/RawSchemaClientFactory.java +++ b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/schema/RawSchemaClientFactory.java @@ -5,6 +5,7 @@ import org.glassfish.jersey.client.ClientProperties; import pl.allegro.tech.hermes.common.config.ConfigFactory; import pl.allegro.tech.hermes.common.config.Configs; +import pl.allegro.tech.hermes.common.metric.HermesMetrics; import pl.allegro.tech.hermes.schema.RawSchemaClient; import pl.allegro.tech.hermes.schema.confluent.SchemaRegistryRawSchemaClient; import pl.allegro.tech.hermes.schema.schemarepo.SchemaRepoRawSchemaClient; @@ -17,10 +18,12 @@ public class RawSchemaClientFactory implements Factory { private final ConfigFactory configFactory; + private final HermesMetrics hermesMetrics; @Inject - public RawSchemaClientFactory(ConfigFactory configFactory) { + public RawSchemaClientFactory(ConfigFactory configFactory, HermesMetrics hermesMetrics) { this.configFactory = configFactory; + this.hermesMetrics = hermesMetrics; } @Override @@ -36,16 +39,21 @@ public RawSchemaClient provide() { String schemaRepositoryType = configFactory.getStringProperty(Configs.SCHEMA_REPOSITORY_TYPE).toUpperCase(); Client client = ClientBuilder.newClient(config); URI schemaRepositoryServerUri = URI.create(configFactory.getStringProperty(Configs.SCHEMA_REPOSITORY_SERVER_URL)); - switch (SchemaRepositoryType.valueOf(schemaRepositoryType)) { + SchemaRepositoryType repoType = SchemaRepositoryType.valueOf(schemaRepositoryType); + switch (repoType) { case SCHEMA_REPO: - return new SchemaRepoRawSchemaClient(client, schemaRepositoryServerUri); + return createMetricsTrackingClient(new SchemaRepoRawSchemaClient(client, schemaRepositoryServerUri), repoType); case SCHEMA_REGISTRY: - return new SchemaRegistryRawSchemaClient(client, schemaRepositoryServerUri); + return createMetricsTrackingClient(new SchemaRegistryRawSchemaClient(client, schemaRepositoryServerUri), repoType); default: throw new IllegalStateException("Unknown schema repository type " + schemaRepositoryType); } } + private RawSchemaClient createMetricsTrackingClient(RawSchemaClient rawSchemaClient, SchemaRepositoryType schemaRepositoryType) { + return new ReadMetricsTrackingRawSchemaClient(rawSchemaClient, hermesMetrics, schemaRepositoryType); + } + @Override public void dispose(RawSchemaClient instance) { diff --git a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/schema/ReadMetricsTrackingRawSchemaClient.java b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/schema/ReadMetricsTrackingRawSchemaClient.java new file mode 100644 index 0000000000..e500108000 --- /dev/null +++ b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/schema/ReadMetricsTrackingRawSchemaClient.java @@ -0,0 +1,72 @@ +package pl.allegro.tech.hermes.common.schema; + +import com.codahale.metrics.Timer; +import pl.allegro.tech.hermes.api.RawSchema; +import pl.allegro.tech.hermes.api.TopicName; +import pl.allegro.tech.hermes.common.metric.HermesMetrics; +import pl.allegro.tech.hermes.common.metric.Timers; +import pl.allegro.tech.hermes.schema.RawSchemaClient; +import pl.allegro.tech.hermes.schema.SchemaVersion; + +import java.util.List; +import java.util.Optional; +import java.util.function.Supplier; + +public class ReadMetricsTrackingRawSchemaClient implements RawSchemaClient { + private final RawSchemaClient rawSchemaClient; + private final HermesMetrics hermesMetrics; + private final SchemaRepositoryType schemaRepoType; + + public ReadMetricsTrackingRawSchemaClient( + RawSchemaClient rawSchemaClient, + HermesMetrics hermesMetrics, + SchemaRepositoryType schemaRepoType) { + this.rawSchemaClient = rawSchemaClient; + this.hermesMetrics = hermesMetrics; + this.schemaRepoType = schemaRepoType; + } + + @Override + public Optional getSchema(TopicName topic, SchemaVersion version) { + return timedSchema(() -> rawSchemaClient.getSchema(topic, version)); + } + + @Override + public Optional getLatestSchema(TopicName topic) { + return timedSchema(() -> rawSchemaClient.getLatestSchema(topic)); + } + + @Override + public List getVersions(TopicName topic) { + return timedVersions(() -> rawSchemaClient.getVersions(topic)); + } + + @Override + public void registerSchema(TopicName topic, RawSchema rawSchema) { + rawSchemaClient.registerSchema(topic, rawSchema); + } + + @Override + public void deleteAllSchemaVersions(TopicName topic) { + rawSchemaClient.deleteAllSchemaVersions(topic); + } + + private T timedSchema(Supplier callable) { + return timed(callable, Timers.GET_SCHEMA_LATENCY); + } + + private T timedVersions(Supplier callable) { + return timed(callable, Timers.GET_SCHEMA_VERSIONS_LATENCY); + } + + private T timed(Supplier callable, String schemaTimer) { + try (Timer.Context time = startLatencyTimer(schemaTimer)) { + return callable.get(); + } + } + + private Timer.Context startLatencyTimer(String schemaReadLatency) { + return hermesMetrics.schemaTimer(schemaReadLatency, schemaRepoType).time(); + } + +} diff --git a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/schema/SchemaRepositoryType.java b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/schema/SchemaRepositoryType.java index 3cb7a80d5d..bcce7c56a6 100644 --- a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/schema/SchemaRepositoryType.java +++ b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/schema/SchemaRepositoryType.java @@ -1,5 +1,16 @@ package pl.allegro.tech.hermes.common.schema; public enum SchemaRepositoryType { - SCHEMA_REPO, SCHEMA_REGISTRY + SCHEMA_REPO("schema-repo"), SCHEMA_REGISTRY("schema-registry"); + + private final String metricName; + + SchemaRepositoryType(String metricName) { + this.metricName = metricName; + } + + @Override + public String toString() { + return metricName; + } } diff --git a/hermes-common/src/test/groovy/pl/allegro/tech/hermes/common/schema/ReadMetricsTrackingRawSchemaClientTest.groovy b/hermes-common/src/test/groovy/pl/allegro/tech/hermes/common/schema/ReadMetricsTrackingRawSchemaClientTest.groovy new file mode 100644 index 0000000000..334c035a53 --- /dev/null +++ b/hermes-common/src/test/groovy/pl/allegro/tech/hermes/common/schema/ReadMetricsTrackingRawSchemaClientTest.groovy @@ -0,0 +1,91 @@ +package pl.allegro.tech.hermes.common.schema + +import com.codahale.metrics.Timer +import pl.allegro.tech.hermes.api.RawSchema +import pl.allegro.tech.hermes.api.TopicName +import pl.allegro.tech.hermes.common.metric.HermesMetrics +import pl.allegro.tech.hermes.common.metric.Timers +import pl.allegro.tech.hermes.schema.RawSchemaClient +import pl.allegro.tech.hermes.schema.SchemaVersion +import spock.lang.Shared +import spock.lang.Specification +import spock.lang.Subject + +import static pl.allegro.tech.hermes.common.schema.SchemaRepositoryType.SCHEMA_REGISTRY + +class ReadMetricsTrackingRawSchemaClientTest extends Specification { + @Shared + TopicName topicName = TopicName.fromQualifiedName("someGroup.someTopic") + + @Shared + SchemaVersion schemaVersion = SchemaVersion.valueOf(1) + + @Shared + RawSchema schema = RawSchema.valueOf("some_schema") + + HermesMetrics hermesMetrics = Mock() + + Timer schemaLatencyTimer = new Timer() + + Timer schemaVersionsLatencyTimer = new Timer() + + RawSchemaClient rawSchemaClient = Mock() + + @Subject + RawSchemaClient readMetricsTrackingClient = new ReadMetricsTrackingRawSchemaClient(rawSchemaClient, hermesMetrics, SCHEMA_REGISTRY) + + def "should track latency metrics for schema retrieval"(){ + expect: + schemaLatencyTimer.count == 0 + + when: + readMetricsTrackingClient.getSchema(topicName, schemaVersion) + + then: + 1 * hermesMetrics.schemaTimer(Timers.GET_SCHEMA_LATENCY, SCHEMA_REGISTRY) >> schemaLatencyTimer + 1 * rawSchemaClient.getSchema(topicName, schemaVersion) + schemaLatencyTimer.count == 1 + } + + def "should track latency metrics for latest schema retrieval"(){ + expect: + schemaLatencyTimer.count == 0 + + when: + readMetricsTrackingClient.getLatestSchema(topicName) + + then: + 1 * hermesMetrics.schemaTimer(Timers.GET_SCHEMA_LATENCY, SCHEMA_REGISTRY) >> schemaLatencyTimer + 1 * rawSchemaClient.getLatestSchema(topicName) + schemaLatencyTimer.count == 1 + } + + def "should track latency metrics for versions retrieval"(){ + expect: + schemaVersionsLatencyTimer.count == 0 + + when: + readMetricsTrackingClient.getVersions(topicName) + + then: + 1 * hermesMetrics.schemaTimer(Timers.GET_SCHEMA_VERSIONS_LATENCY, SCHEMA_REGISTRY) >> schemaVersionsLatencyTimer + 1 * rawSchemaClient.getVersions(topicName) + schemaVersionsLatencyTimer.count == 1 + } + + def "should call inner client for non-read operations"() { + when: + readMetricsTrackingClient.deleteAllSchemaVersions(topicName) + + then: + 1 * rawSchemaClient.deleteAllSchemaVersions(topicName) + schemaVersionsLatencyTimer.count == 0 + + when: + readMetricsTrackingClient.registerSchema(topicName, schema) + + then: + 1 * rawSchemaClient.registerSchema(topicName, schema) + schemaLatencyTimer.count == 0 + } +} diff --git a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/buffer/BackupMessagesLoader.java b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/buffer/BackupMessagesLoader.java index ab53e2d877..da388b776d 100644 --- a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/buffer/BackupMessagesLoader.java +++ b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/buffer/BackupMessagesLoader.java @@ -10,7 +10,7 @@ import pl.allegro.tech.hermes.frontend.cache.topic.TopicsCache; import pl.allegro.tech.hermes.frontend.listeners.BrokerListeners; import pl.allegro.tech.hermes.frontend.metric.CachedTopic; -import pl.allegro.tech.hermes.frontend.metric.StartedTimersPair; +import pl.allegro.tech.hermes.common.metric.timer.StartedTimersPair; import pl.allegro.tech.hermes.frontend.producer.BrokerMessageProducer; import pl.allegro.tech.hermes.frontend.publishing.PublishingCallback; import pl.allegro.tech.hermes.frontend.publishing.message.JsonMessage; @@ -26,17 +26,10 @@ import java.util.List; import java.util.Optional; import java.util.Set; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Executor; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicReference; -import static java.util.concurrent.TimeUnit.SECONDS; import static pl.allegro.tech.hermes.common.config.Configs.*; public class BackupMessagesLoader { diff --git a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/metric/CachedTopic.java b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/metric/CachedTopic.java index 1151d1ca1b..f122f4d3ab 100644 --- a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/metric/CachedTopic.java +++ b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/metric/CachedTopic.java @@ -12,6 +12,7 @@ import pl.allegro.tech.hermes.common.metric.HermesMetrics; import pl.allegro.tech.hermes.common.metric.Meters; import pl.allegro.tech.hermes.common.metric.Timers; +import pl.allegro.tech.hermes.common.metric.timer.StartedTimersPair; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; diff --git a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/ExchangeMetrics.java b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/ExchangeMetrics.java index 1a86532678..11bd41b2e3 100644 --- a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/ExchangeMetrics.java +++ b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/ExchangeMetrics.java @@ -4,7 +4,7 @@ import io.undertow.server.HttpServerExchange; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import pl.allegro.tech.hermes.frontend.metric.StartedTimersPair; +import pl.allegro.tech.hermes.common.metric.timer.StartedTimersPair; import pl.allegro.tech.hermes.frontend.metric.CachedTopic; class ExchangeMetrics implements ExchangeCompletionListener { diff --git a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/MessageReadHandler.java b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/MessageReadHandler.java index 7567d4b33b..d6c09815bc 100644 --- a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/MessageReadHandler.java +++ b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/MessageReadHandler.java @@ -6,7 +6,7 @@ import io.undertow.server.HttpServerExchange; import pl.allegro.tech.hermes.common.config.ConfigFactory; import pl.allegro.tech.hermes.common.config.Configs; -import pl.allegro.tech.hermes.frontend.metric.StartedTimersPair; +import pl.allegro.tech.hermes.common.metric.timer.StartedTimersPair; import pl.allegro.tech.hermes.frontend.publishing.handlers.end.MessageErrorProcessor; import pl.allegro.tech.hermes.frontend.publishing.message.MessageState; diff --git a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/PublishingHandler.java b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/PublishingHandler.java index 01f5ef6a95..724977e91c 100644 --- a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/PublishingHandler.java +++ b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/PublishingHandler.java @@ -3,7 +3,7 @@ import io.undertow.server.HttpHandler; import io.undertow.server.HttpServerExchange; import pl.allegro.tech.hermes.api.Topic; -import pl.allegro.tech.hermes.frontend.metric.StartedTimersPair; +import pl.allegro.tech.hermes.common.metric.timer.StartedTimersPair; import pl.allegro.tech.hermes.frontend.producer.BrokerMessageProducer; import pl.allegro.tech.hermes.frontend.publishing.PublishingCallback; import pl.allegro.tech.hermes.frontend.publishing.handlers.end.MessageEndProcessor; diff --git a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageFactory.java b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageFactory.java index 061d6b39f0..e5638f5720 100644 --- a/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageFactory.java +++ b/hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageFactory.java @@ -9,7 +9,7 @@ import pl.allegro.tech.hermes.common.http.MessageMetadataHeaders; import pl.allegro.tech.hermes.common.message.wrapper.MessageContentWrapper; import pl.allegro.tech.hermes.common.message.wrapper.UnsupportedContentTypeException; -import pl.allegro.tech.hermes.frontend.metric.StartedTimersPair; +import pl.allegro.tech.hermes.common.metric.timer.StartedTimersPair; import pl.allegro.tech.hermes.common.message.wrapper.WrappingException; import pl.allegro.tech.hermes.schema.SchemaRepository; import pl.allegro.tech.hermes.frontend.publishing.avro.AvroMessage; diff --git a/hermes-frontend/src/test/java/pl/allegro/tech/hermes/frontend/buffer/BackupMessagesLoaderTest.java b/hermes-frontend/src/test/java/pl/allegro/tech/hermes/frontend/buffer/BackupMessagesLoaderTest.java index bc92e426c8..bf8a015584 100644 --- a/hermes-frontend/src/test/java/pl/allegro/tech/hermes/frontend/buffer/BackupMessagesLoaderTest.java +++ b/hermes-frontend/src/test/java/pl/allegro/tech/hermes/frontend/buffer/BackupMessagesLoaderTest.java @@ -13,7 +13,7 @@ import pl.allegro.tech.hermes.frontend.cache.topic.TopicsCache; import pl.allegro.tech.hermes.frontend.listeners.BrokerListeners; import pl.allegro.tech.hermes.frontend.metric.CachedTopic; -import pl.allegro.tech.hermes.frontend.metric.StartedTimersPair; +import pl.allegro.tech.hermes.common.metric.timer.StartedTimersPair; import pl.allegro.tech.hermes.frontend.producer.BrokerMessageProducer; import pl.allegro.tech.hermes.frontend.publishing.PublishingCallback; import pl.allegro.tech.hermes.frontend.publishing.message.JsonMessage; diff --git a/hermes-metrics/src/main/java/pl/allegro/tech/hermes/metrics/PathContext.java b/hermes-metrics/src/main/java/pl/allegro/tech/hermes/metrics/PathContext.java index 0f2fa45de0..49ac537c38 100644 --- a/hermes-metrics/src/main/java/pl/allegro/tech/hermes/metrics/PathContext.java +++ b/hermes-metrics/src/main/java/pl/allegro/tech/hermes/metrics/PathContext.java @@ -14,6 +14,7 @@ public class PathContext { private final Optional httpCodeFamily; private final Optional executorName; private final Optional oAuthProviderName; + private final Optional schemaRepoType; private PathContext(Optional group, Optional topic, @@ -24,7 +25,8 @@ private PathContext(Optional group, Optional httpCode, Optional httpCodeFamily, Optional executorName, - Optional oAuthProviderName) { + Optional oAuthProviderName, + Optional schemaRepoType) { this.group = group; this.topic = topic; this.subscription = subscription; @@ -35,6 +37,7 @@ private PathContext(Optional group, this.httpCodeFamily = httpCodeFamily; this.executorName = executorName; this.oAuthProviderName = oAuthProviderName; + this.schemaRepoType = schemaRepoType; } public Optional getGroup() { @@ -77,6 +80,10 @@ public Optional getoAuthProviderName() { return oAuthProviderName; } + public Optional getSchemaRepoType() { + return schemaRepoType; + } + public static Builder pathContext() { return new Builder(); } @@ -93,6 +100,7 @@ public static class Builder { private Optional httpCodeFamily = Optional.empty(); private Optional executorName = Optional.empty(); private Optional oAuthProviderName = Optional.empty(); + private Optional schemaRepoType = Optional.empty(); public Builder withGroup(String group) { this.group = Optional.of(group); @@ -143,10 +151,14 @@ public Builder withOAuthProvider(String oAuthProviderName) { this.oAuthProviderName = Optional.of(oAuthProviderName); return this; } + public Builder withSchemaRepoType(String schemaRepoType) { + this.schemaRepoType = Optional.of(schemaRepoType); + return this; + } public PathContext build() { return new PathContext(group, topic, subscription, kafkaTopic, partition, kafkaCluster, - httpCode, httpCodeFamily, executorName, oAuthProviderName); + httpCode, httpCodeFamily, executorName, oAuthProviderName, schemaRepoType); } } } diff --git a/hermes-metrics/src/main/java/pl/allegro/tech/hermes/metrics/PathsCompiler.java b/hermes-metrics/src/main/java/pl/allegro/tech/hermes/metrics/PathsCompiler.java index fecc087e42..a704a0482a 100644 --- a/hermes-metrics/src/main/java/pl/allegro/tech/hermes/metrics/PathsCompiler.java +++ b/hermes-metrics/src/main/java/pl/allegro/tech/hermes/metrics/PathsCompiler.java @@ -17,6 +17,7 @@ public class PathsCompiler { public static final String HTTP_CODE_FAMILY = "$http_family_of_code"; public static final String EXECUTOR_NAME = "$executor_name"; public static final String OAUTH_PROVIDER_NAME = "$oauth_provider_name"; + public static final String SCHEMA_REPO_TYPE = "$schema_repo_type"; private final String hostname; @@ -41,6 +42,7 @@ public String compile(String path, PathContext context) { context.getHttpCodeFamily().ifPresent(cf -> pathBuilder.replaceAll(HTTP_CODE_FAMILY, cf)); context.getExecutorName().ifPresent(c -> pathBuilder.replaceAll(EXECUTOR_NAME, c)); context.getoAuthProviderName().ifPresent(c -> pathBuilder.replaceAll(OAUTH_PROVIDER_NAME, c)); + context.getSchemaRepoType().ifPresent(c -> pathBuilder.replaceAll(SCHEMA_REPO_TYPE, c)); pathBuilder.replaceAll(HOSTNAME, hostname); From 4557cd36cecc32c3296f57179d12de9e1e7b5a04 Mon Sep 17 00:00:00 2001 From: Lukasz Druminski Date: Wed, 15 Mar 2017 14:07:57 +0100 Subject: [PATCH 4/4] Updating changelog about 0.11.2 release --- CHANGELOG.md | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1bd11bb877..c66e180762 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,23 @@ +## 0.11.2 (15.03.2017) + +### Enhancements + +#### ([738](https://github.com/allegro/hermes/pull/738)) Payload content-type check and error handling + +From now on, clients who send HTTP request without specified `Content-Type` on an Avro topic will receive proper error message. + +#### ([739](https://github.com/allegro/hermes/pull/739)) Added latency metrics for schema registry + +Metrics are available in the following path: + +``` +schema..latency.read-schema +``` + +### Bugfixes + +#### ([740](https://github.com/allegro/hermes/issues/740)) Invalid metrics names in zookeeper for topics with underscore in name + ## 0.11.1 (7.03.2017) ### Features