From 1b1c617ee5e9f83f1fa496f0a3a31bfd65d04f13 Mon Sep 17 00:00:00 2001 From: aayustark007-fk Date: Fri, 7 Jun 2024 10:30:12 +0530 Subject: [PATCH 1/7] add initial message delivery implementation --- .../consumer/delivery/DeliveryResponse.java | 6 ++ .../consumer/delivery/MessageDelivery.java | 71 +++++++++++++++++++ consumer/src/main/java/module-info.java | 1 + .../flipkart/varadhi/entities/Endpoint.java | 6 +- .../com/flipkart/varadhi/IamPolicyTests.java | 16 +++++ 5 files changed, 98 insertions(+), 2 deletions(-) create mode 100644 consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/DeliveryResponse.java create mode 100644 consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java create mode 100644 server/src/testE2E/java/com/flipkart/varadhi/IamPolicyTests.java diff --git a/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/DeliveryResponse.java b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/DeliveryResponse.java new file mode 100644 index 00000000..99e56403 --- /dev/null +++ b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/DeliveryResponse.java @@ -0,0 +1,6 @@ +package com.flipkart.varadhi.consumer.delivery; + +import com.flipkart.varadhi.entities.Endpoint; + +public record DeliveryResponse(int statusCode, Endpoint.Protocol protocol, byte[] body) { +} diff --git a/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java new file mode 100644 index 00000000..029c8cc0 --- /dev/null +++ b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java @@ -0,0 +1,71 @@ +package com.flipkart.varadhi.consumer.delivery; + +import com.flipkart.varadhi.consumer.MessageTracker; +import com.flipkart.varadhi.entities.Endpoint; +import com.flipkart.varadhi.exceptions.NotImplementedException; +import com.google.common.collect.Multimap; + +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; + +public interface MessageDelivery { + static MessageDelivery of(Endpoint endpoint) { + return switch (endpoint.getProtocol()) { + case HTTP1_1 -> new HttpMessageDelivery(endpoint); + case HTTP2 -> throw new NotImplementedException("HTTP2 is not supported yet"); + default -> throw new IllegalArgumentException("Unsupported protocol: " + endpoint.getProtocol()); + }; + } + + CompletableFuture deliver(MessageTracker messageTracker) + throws Exception; // or should be PolledMessage? + + class HttpMessageDelivery implements MessageDelivery { + private final Endpoint.HttpEndpoint endpoint; + private final HttpClient httpClient; + + public HttpMessageDelivery(Endpoint endpoint) { + this.endpoint = (Endpoint.HttpEndpoint) endpoint; + this.httpClient = HttpClient.newBuilder() + .version(this.endpoint.isHttp2Supported() ? HttpClient.Version.HTTP_2 : HttpClient.Version.HTTP_1_1) + .connectTimeout(Duration.ofMillis(this.endpoint.getConnectTimeoutMs())) + .followRedirects(HttpClient.Redirect.NORMAL) + .build(); + // TODO(aayush): SSL/TLS support? Auth support? + } + + @Override + public CompletableFuture deliver(MessageTracker messageTracker) throws Exception { + // TODO(aayush): PolledMessage getPayload, getHeaders support required + HttpRequest.Builder requestBuilder = HttpRequest.newBuilder() + .uri(endpoint.getUrl().toURI()) + .timeout(Duration.ofMillis(endpoint.getRequestTimeoutMs())) + .header("Content-Type", endpoint.getContentType()) + .method( + endpoint.getMethod(), + HttpRequest.BodyPublishers.ofByteArray(messageTracker.getMessage().getPayload()) + ); + + // apply request headers from message + Multimap requestHeaders = messageTracker.getMessage().getRequestHeaders(); + if (requestHeaders != null) { + requestHeaders.entries().forEach(entry -> requestBuilder.header(entry.getKey(), entry.getValue())); + } + + HttpRequest request = requestBuilder.build(); + + return httpClient.sendAsync( + request, HttpResponse.BodyHandlers.ofByteArray()) // TODO(aayush): response as string or byte array? + .thenApply(response -> new DeliveryResponse(response.statusCode(), endpoint.getProtocol(), + response.body() + )) + .exceptionally(e -> { + // log error + return new DeliveryResponse(500, endpoint.getProtocol(), e.getMessage().getBytes()); + }); + } + } +} diff --git a/consumer/src/main/java/module-info.java b/consumer/src/main/java/module-info.java index b1bbba2f..cdf85d6b 100644 --- a/consumer/src/main/java/module-info.java +++ b/consumer/src/main/java/module-info.java @@ -12,4 +12,5 @@ requires com.flipkart.varadhi.core; requires jakarta.annotation; requires com.fasterxml.jackson.annotation; + requires java.net.http; } diff --git a/entities/src/main/java/com/flipkart/varadhi/entities/Endpoint.java b/entities/src/main/java/com/flipkart/varadhi/entities/Endpoint.java index 582bd22b..fa115cf4 100644 --- a/entities/src/main/java/com/flipkart/varadhi/entities/Endpoint.java +++ b/entities/src/main/java/com/flipkart/varadhi/entities/Endpoint.java @@ -1,5 +1,6 @@ package com.flipkart.varadhi.entities; +import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.annotation.JsonSubTypes; import com.fasterxml.jackson.annotation.JsonTypeInfo; import lombok.Data; @@ -16,7 +17,8 @@ }) public abstract sealed class Endpoint { - abstract Protocol getProtocol(); + @JsonIgnore + public abstract Protocol getProtocol(); public enum Protocol { HTTP1_1, @@ -35,7 +37,7 @@ public static final class HttpEndpoint extends Endpoint { private final boolean http2Supported; @Override - Protocol getProtocol() { + public Protocol getProtocol() { return http2Supported ? Protocol.HTTP2 : Protocol.HTTP1_1; } } diff --git a/server/src/testE2E/java/com/flipkart/varadhi/IamPolicyTests.java b/server/src/testE2E/java/com/flipkart/varadhi/IamPolicyTests.java new file mode 100644 index 00000000..7a998c62 --- /dev/null +++ b/server/src/testE2E/java/com/flipkart/varadhi/IamPolicyTests.java @@ -0,0 +1,16 @@ +package com.flipkart.varadhi; + +public class IamPolicyTests extends E2EBase { + + /** + * Tests for iam policy, some cases to test are: + * - initial superuser should have all permissions + * - thanos creates flipkart org + * - create admin user via set policy + * - admin user should be able to create more user roles + * - non admin users should not be able to create roles + * - admin user can create team and project + * - similarly for team and project scoped roles, do the same + * - all the way down to topic level and produce level + */ +} From 26ab68266adcd3641780db48186165578fae1def Mon Sep 17 00:00:00 2001 From: aayustark007-fk Date: Fri, 7 Jun 2024 22:13:59 +0530 Subject: [PATCH 2/7] cleanup code --- .../consumer/delivery/MessageDelivery.java | 16 +++++++--------- .../com/flipkart/varadhi/IamPolicyTests.java | 16 ---------------- 2 files changed, 7 insertions(+), 25 deletions(-) delete mode 100644 server/src/testE2E/java/com/flipkart/varadhi/IamPolicyTests.java diff --git a/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java index 029c8cc0..78a54cf6 100644 --- a/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java +++ b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java @@ -1,7 +1,7 @@ package com.flipkart.varadhi.consumer.delivery; -import com.flipkart.varadhi.consumer.MessageTracker; import com.flipkart.varadhi.entities.Endpoint; +import com.flipkart.varadhi.entities.Message; import com.flipkart.varadhi.exceptions.NotImplementedException; import com.google.common.collect.Multimap; @@ -20,8 +20,8 @@ static MessageDelivery of(Endpoint endpoint) { }; } - CompletableFuture deliver(MessageTracker messageTracker) - throws Exception; // or should be PolledMessage? + CompletableFuture deliver(Message message) + throws Exception; class HttpMessageDelivery implements MessageDelivery { private final Endpoint.HttpEndpoint endpoint; @@ -34,23 +34,21 @@ public HttpMessageDelivery(Endpoint endpoint) { .connectTimeout(Duration.ofMillis(this.endpoint.getConnectTimeoutMs())) .followRedirects(HttpClient.Redirect.NORMAL) .build(); - // TODO(aayush): SSL/TLS support? Auth support? } @Override - public CompletableFuture deliver(MessageTracker messageTracker) throws Exception { - // TODO(aayush): PolledMessage getPayload, getHeaders support required + public CompletableFuture deliver(Message message) throws Exception { HttpRequest.Builder requestBuilder = HttpRequest.newBuilder() .uri(endpoint.getUrl().toURI()) .timeout(Duration.ofMillis(endpoint.getRequestTimeoutMs())) .header("Content-Type", endpoint.getContentType()) .method( endpoint.getMethod(), - HttpRequest.BodyPublishers.ofByteArray(messageTracker.getMessage().getPayload()) + HttpRequest.BodyPublishers.ofByteArray(message.getPayload()) ); // apply request headers from message - Multimap requestHeaders = messageTracker.getMessage().getRequestHeaders(); + Multimap requestHeaders = message.getRequestHeaders(); if (requestHeaders != null) { requestHeaders.entries().forEach(entry -> requestBuilder.header(entry.getKey(), entry.getValue())); } @@ -58,7 +56,7 @@ public CompletableFuture deliver(MessageTracker messageTracker HttpRequest request = requestBuilder.build(); return httpClient.sendAsync( - request, HttpResponse.BodyHandlers.ofByteArray()) // TODO(aayush): response as string or byte array? + request, HttpResponse.BodyHandlers.ofByteArray()) .thenApply(response -> new DeliveryResponse(response.statusCode(), endpoint.getProtocol(), response.body() )) diff --git a/server/src/testE2E/java/com/flipkart/varadhi/IamPolicyTests.java b/server/src/testE2E/java/com/flipkart/varadhi/IamPolicyTests.java deleted file mode 100644 index 7a998c62..00000000 --- a/server/src/testE2E/java/com/flipkart/varadhi/IamPolicyTests.java +++ /dev/null @@ -1,16 +0,0 @@ -package com.flipkart.varadhi; - -public class IamPolicyTests extends E2EBase { - - /** - * Tests for iam policy, some cases to test are: - * - initial superuser should have all permissions - * - thanos creates flipkart org - * - create admin user via set policy - * - admin user should be able to create more user roles - * - non admin users should not be able to create roles - * - admin user can create team and project - * - similarly for team and project scoped roles, do the same - * - all the way down to topic level and produce level - */ -} From 9f56510758331b3ecb9e45b1f7d39806d43b98f4 Mon Sep 17 00:00:00 2001 From: aayustark007-fk Date: Wed, 12 Jun 2024 20:42:20 +0530 Subject: [PATCH 3/7] ensure empty body adds no publisher some endpoints will return error code if GET or DELETE methods are called with a body Signed-off-by: aayustark007-fk --- .../flipkart/varadhi/consumer/delivery/MessageDelivery.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java index 78a54cf6..6d5e0f0f 100644 --- a/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java +++ b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java @@ -4,6 +4,7 @@ import com.flipkart.varadhi.entities.Message; import com.flipkart.varadhi.exceptions.NotImplementedException; import com.google.common.collect.Multimap; +import org.apache.commons.lang3.ArrayUtils; import java.net.http.HttpClient; import java.net.http.HttpRequest; @@ -44,7 +45,8 @@ public CompletableFuture deliver(Message message) throws Excep .header("Content-Type", endpoint.getContentType()) .method( endpoint.getMethod(), - HttpRequest.BodyPublishers.ofByteArray(message.getPayload()) + ArrayUtils.isEmpty(message.getPayload()) ? HttpRequest.BodyPublishers.noBody() : + HttpRequest.BodyPublishers.ofByteArray(message.getPayload()) ); // apply request headers from message From 005107bc91137e9db7e3a1b1cb14528a787d4ec6 Mon Sep 17 00:00:00 2001 From: aayustark007-fk Date: Wed, 19 Jun 2024 12:12:56 +0530 Subject: [PATCH 4/7] using protocol as sub type property for Endpoint --- .../main/java/com/flipkart/varadhi/entities/Endpoint.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/entities/src/main/java/com/flipkart/varadhi/entities/Endpoint.java b/entities/src/main/java/com/flipkart/varadhi/entities/Endpoint.java index fa115cf4..20f2a0dc 100644 --- a/entities/src/main/java/com/flipkart/varadhi/entities/Endpoint.java +++ b/entities/src/main/java/com/flipkart/varadhi/entities/Endpoint.java @@ -10,10 +10,11 @@ @JsonTypeInfo( use = JsonTypeInfo.Id.NAME, - property = "@endpointType" + property = "protocol" ) @JsonSubTypes({ - @JsonSubTypes.Type(Endpoint.HttpEndpoint.class), + @JsonSubTypes.Type(value = Endpoint.HttpEndpoint.class, name = "HTTP1_1"), + @JsonSubTypes.Type(value = Endpoint.HttpEndpoint.class, name = "HTTP2"), }) public abstract sealed class Endpoint { From b154c1ad4702fd940595913335e28afc32b53891 Mon Sep 17 00:00:00 2001 From: aayustark007-fk Date: Fri, 21 Jun 2024 10:12:59 +0530 Subject: [PATCH 5/7] pr comments addressed --- .../varadhi/consumer/delivery/MessageDelivery.java | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java index 6d5e0f0f..66331c23 100644 --- a/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java +++ b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java @@ -15,7 +15,7 @@ public interface MessageDelivery { static MessageDelivery of(Endpoint endpoint) { return switch (endpoint.getProtocol()) { - case HTTP1_1 -> new HttpMessageDelivery(endpoint); + case HTTP1_1 -> new HttpMessageDelivery((Endpoint.HttpEndpoint) endpoint); case HTTP2 -> throw new NotImplementedException("HTTP2 is not supported yet"); default -> throw new IllegalArgumentException("Unsupported protocol: " + endpoint.getProtocol()); }; @@ -28,8 +28,8 @@ class HttpMessageDelivery implements MessageDelivery { private final Endpoint.HttpEndpoint endpoint; private final HttpClient httpClient; - public HttpMessageDelivery(Endpoint endpoint) { - this.endpoint = (Endpoint.HttpEndpoint) endpoint; + public HttpMessageDelivery(Endpoint.HttpEndpoint endpoint) { + this.endpoint = endpoint; this.httpClient = HttpClient.newBuilder() .version(this.endpoint.isHttp2Supported() ? HttpClient.Version.HTTP_2 : HttpClient.Version.HTTP_1_1) .connectTimeout(Duration.ofMillis(this.endpoint.getConnectTimeoutMs())) @@ -61,11 +61,7 @@ public CompletableFuture deliver(Message message) throws Excep request, HttpResponse.BodyHandlers.ofByteArray()) .thenApply(response -> new DeliveryResponse(response.statusCode(), endpoint.getProtocol(), response.body() - )) - .exceptionally(e -> { - // log error - return new DeliveryResponse(500, endpoint.getProtocol(), e.getMessage().getBytes()); - }); + )); } } } From 9f552e2b9da7d3b35be7c93b0ef387bdfebbd606 Mon Sep 17 00:00:00 2001 From: aayustark007-fk Date: Sun, 23 Jun 2024 19:04:27 +0530 Subject: [PATCH 6/7] add http client supplier to message delivery --- .../consumer/delivery/MessageDelivery.java | 15 +++++---------- 1 file changed, 5 insertions(+), 10 deletions(-) diff --git a/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java index 66331c23..1aca4472 100644 --- a/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java +++ b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java @@ -2,7 +2,6 @@ import com.flipkart.varadhi.entities.Endpoint; import com.flipkart.varadhi.entities.Message; -import com.flipkart.varadhi.exceptions.NotImplementedException; import com.google.common.collect.Multimap; import org.apache.commons.lang3.ArrayUtils; @@ -11,12 +10,12 @@ import java.net.http.HttpResponse; import java.time.Duration; import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; public interface MessageDelivery { - static MessageDelivery of(Endpoint endpoint) { + static MessageDelivery of(Endpoint endpoint, Supplier httpClientSupplier) { return switch (endpoint.getProtocol()) { - case HTTP1_1 -> new HttpMessageDelivery((Endpoint.HttpEndpoint) endpoint); - case HTTP2 -> throw new NotImplementedException("HTTP2 is not supported yet"); + case HTTP1_1, HTTP2 -> new HttpMessageDelivery((Endpoint.HttpEndpoint) endpoint, httpClientSupplier.get()); default -> throw new IllegalArgumentException("Unsupported protocol: " + endpoint.getProtocol()); }; } @@ -28,13 +27,9 @@ class HttpMessageDelivery implements MessageDelivery { private final Endpoint.HttpEndpoint endpoint; private final HttpClient httpClient; - public HttpMessageDelivery(Endpoint.HttpEndpoint endpoint) { + public HttpMessageDelivery(Endpoint.HttpEndpoint endpoint, HttpClient client) { this.endpoint = endpoint; - this.httpClient = HttpClient.newBuilder() - .version(this.endpoint.isHttp2Supported() ? HttpClient.Version.HTTP_2 : HttpClient.Version.HTTP_1_1) - .connectTimeout(Duration.ofMillis(this.endpoint.getConnectTimeoutMs())) - .followRedirects(HttpClient.Redirect.NORMAL) - .build(); + this.httpClient = client; } @Override From b46549946688983d1ea407bcad2a4a63782b952f Mon Sep 17 00:00:00 2001 From: aayustark007-fk Date: Sun, 23 Jun 2024 19:13:08 +0530 Subject: [PATCH 7/7] refactor Http Endpoint class URL to URI for better usablility with HttpClients --- .../consumer/delivery/MessageDelivery.java | 2 +- .../com/flipkart/varadhi/entities/Endpoint.java | 4 ++-- .../services/SubscriptionServiceTest.java | 8 ++------ .../web/admin/SubscriptionHandlersTest.java | 16 ++++------------ .../com/flipkart/varadhi/SubscriptionTests.java | 14 +++----------- 5 files changed, 12 insertions(+), 32 deletions(-) diff --git a/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java index 1aca4472..9842beb1 100644 --- a/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java +++ b/consumer/src/main/java/com/flipkart/varadhi/consumer/delivery/MessageDelivery.java @@ -35,7 +35,7 @@ public HttpMessageDelivery(Endpoint.HttpEndpoint endpoint, HttpClient client) { @Override public CompletableFuture deliver(Message message) throws Exception { HttpRequest.Builder requestBuilder = HttpRequest.newBuilder() - .uri(endpoint.getUrl().toURI()) + .uri(endpoint.getUri()) .timeout(Duration.ofMillis(endpoint.getRequestTimeoutMs())) .header("Content-Type", endpoint.getContentType()) .method( diff --git a/entities/src/main/java/com/flipkart/varadhi/entities/Endpoint.java b/entities/src/main/java/com/flipkart/varadhi/entities/Endpoint.java index 20f2a0dc..d5b26ae4 100644 --- a/entities/src/main/java/com/flipkart/varadhi/entities/Endpoint.java +++ b/entities/src/main/java/com/flipkart/varadhi/entities/Endpoint.java @@ -6,7 +6,7 @@ import lombok.Data; import lombok.EqualsAndHashCode; -import java.net.URL; +import java.net.URI; @JsonTypeInfo( use = JsonTypeInfo.Id.NAME, @@ -29,7 +29,7 @@ public enum Protocol { @EqualsAndHashCode(callSuper = true) @Data public static final class HttpEndpoint extends Endpoint { - private final URL url; + private final URI uri; private final String method; private final String contentType; private final long connectTimeoutMs; diff --git a/server/src/test/java/com/flipkart/varadhi/services/SubscriptionServiceTest.java b/server/src/test/java/com/flipkart/varadhi/services/SubscriptionServiceTest.java index 88f753f5..6f0ff659 100644 --- a/server/src/test/java/com/flipkart/varadhi/services/SubscriptionServiceTest.java +++ b/server/src/test/java/com/flipkart/varadhi/services/SubscriptionServiceTest.java @@ -39,7 +39,7 @@ import org.junit.jupiter.api.extension.ExtendWith; import java.net.MalformedURLException; -import java.net.URL; +import java.net.URI; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -115,11 +115,7 @@ void testSubscriptionEntitiesSerDe() { 1, 1, 1, 1 ); ConsumptionPolicy consumptionPolicy = new ConsumptionPolicy(1, 1, false, 1, null); - try { - endpoint = new Endpoint.HttpEndpoint(new URL("http", "localhost", "hello"), "GET", "", 500, 500, false); - } catch (MalformedURLException e) { - throw new RuntimeException(e); - } + endpoint = new Endpoint.HttpEndpoint(URI.create("http://localhost:8080"), "GET", "", 500, 500, false); TopicCapacityPolicy capacity = Constants.DefaultTopicCapacity; String region = "default"; diff --git a/server/src/test/java/com/flipkart/varadhi/web/admin/SubscriptionHandlersTest.java b/server/src/test/java/com/flipkart/varadhi/web/admin/SubscriptionHandlersTest.java index 3880ca8e..f8b4e0a8 100644 --- a/server/src/test/java/com/flipkart/varadhi/web/admin/SubscriptionHandlersTest.java +++ b/server/src/test/java/com/flipkart/varadhi/web/admin/SubscriptionHandlersTest.java @@ -1,10 +1,10 @@ package com.flipkart.varadhi.web.admin; -import com.flipkart.varadhi.services.VaradhiTopicService; import com.flipkart.varadhi.entities.*; import com.flipkart.varadhi.exceptions.ResourceNotFoundException; import com.flipkart.varadhi.services.ProjectService; import com.flipkart.varadhi.services.SubscriptionService; +import com.flipkart.varadhi.services.VaradhiTopicService; import com.flipkart.varadhi.utils.VaradhiSubscriptionFactory; import com.flipkart.varadhi.web.ErrorResponse; import com.flipkart.varadhi.web.WebTestBase; @@ -18,8 +18,7 @@ import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; -import java.net.MalformedURLException; -import java.net.URL; +import java.net.URI; import java.util.List; import java.util.UUID; import java.util.concurrent.CompletableFuture; @@ -29,7 +28,8 @@ import static org.mockito.Mockito.*; public class SubscriptionHandlersTest extends WebTestBase { - private static final Endpoint endpoint; + private static final Endpoint endpoint = + new Endpoint.HttpEndpoint(URI.create("http://localhost:8080"), "GET", "", 500, 500, false); private static final RetryPolicy retryPolicy = new RetryPolicy( new CodeRange[]{new CodeRange(500, 502)}, RetryPolicy.BackoffType.LINEAR, @@ -39,14 +39,6 @@ public class SubscriptionHandlersTest extends WebTestBase { private static final TopicCapacityPolicy capacityPolicy = new TopicCapacityPolicy(1, 10, 1); private static final SubscriptionShards shards = new SubscriptionUnitShard(0, capacityPolicy, null, null, null); - static { - try { - endpoint = new Endpoint.HttpEndpoint(new URL("http", "localhost", "hello"), "GET", "", 500, 500, false); - } catch (MalformedURLException e) { - throw new RuntimeException(e); - } - } - private final Project project = new Project("project1", 0, "", "team1", "org1"); private final TopicResource topicResource = new TopicResource("topic1", 0, "project2", false, null); SubscriptionHandlers subscriptionHandlers; diff --git a/server/src/testE2E/java/com/flipkart/varadhi/SubscriptionTests.java b/server/src/testE2E/java/com/flipkart/varadhi/SubscriptionTests.java index 77d220f9..6c6dd1d2 100644 --- a/server/src/testE2E/java/com/flipkart/varadhi/SubscriptionTests.java +++ b/server/src/testE2E/java/com/flipkart/varadhi/SubscriptionTests.java @@ -5,8 +5,7 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; -import java.net.MalformedURLException; -import java.net.URL; +import java.net.URI; import java.util.List; import static com.flipkart.varadhi.entities.VersionedEntity.INITIAL_VERSION; @@ -14,21 +13,14 @@ public class SubscriptionTests extends E2EBase { - private static final Endpoint endpoint; + private static final Endpoint endpoint = + new Endpoint.HttpEndpoint(URI.create("http://localhost:8080"), "GET", "", 500, 500, false); private static Org o1; private static Team o1t1; private static Project o1t1p1; private static TopicResource p1t1; private static TopicResource p1t2; - static { - try { - endpoint = new Endpoint.HttpEndpoint(new URL("http", "localhost", "hello"), "GET", "", 500, 500, false); - } catch (MalformedURLException e) { - throw new RuntimeException(e); - } - } - private final RetryPolicy retryPolicy = new RetryPolicy( new CodeRange[]{new CodeRange(500, 502)}, RetryPolicy.BackoffType.LINEAR,