-
Notifications
You must be signed in to change notification settings - Fork 1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix: make endpoints available while waiting for precondition (#5069)
- Loading branch information
Showing
4 changed files
with
229 additions
and
38 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
97 changes: 97 additions & 0 deletions
97
...rest-app/src/test/java/io/confluent/ksql/rest/integration/PreconditionFunctionalTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,97 @@ | ||
| /* | ||
| * Copyright 2018 Confluent Inc. | ||
| * | ||
| * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT | ||
| * WARRANTIES OF ANY KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations under the License. | ||
| */ | ||
|
|
||
| package io.confluent.ksql.rest.integration; | ||
|
|
||
| import static org.hamcrest.MatcherAssert.assertThat; | ||
| import static org.hamcrest.Matchers.is; | ||
|
|
||
| import io.confluent.common.utils.IntegrationTest; | ||
| import io.confluent.ksql.integration.IntegrationTestHarness; | ||
| import io.confluent.ksql.integration.Retry; | ||
| import io.confluent.ksql.rest.entity.KsqlErrorMessage; | ||
| import io.confluent.ksql.rest.server.KsqlRestConfig; | ||
| import io.confluent.ksql.rest.server.KsqlServerPrecondition; | ||
| import io.confluent.ksql.rest.server.TestKsqlRestAppWaitingOnPrecondition; | ||
| import io.confluent.ksql.services.ServiceContext; | ||
| import java.util.Optional; | ||
| import java.util.concurrent.CountDownLatch; | ||
| import java.util.concurrent.TimeUnit; | ||
| import kafka.zookeeper.ZooKeeperClientException; | ||
| import org.junit.BeforeClass; | ||
| import org.junit.ClassRule; | ||
| import org.junit.Test; | ||
| import org.junit.experimental.categories.Category; | ||
| import org.junit.rules.RuleChain; | ||
|
|
||
| @Category({IntegrationTest.class}) | ||
| public class PreconditionFunctionalTest { | ||
|
|
||
| private static final String SERVER_PRECONDITIONS_CONFIG = "ksql.server.preconditions"; | ||
|
|
||
| private static final int CUSTOM_ERROR_CODE = 50370; | ||
| private static final CountDownLatch latch = new CountDownLatch(1); | ||
|
|
||
| private static final IntegrationTestHarness TEST_HARNESS = IntegrationTestHarness.build(); | ||
|
|
||
| private static final TestKsqlRestAppWaitingOnPrecondition REST_APP = TestKsqlRestAppWaitingOnPrecondition | ||
| .builder(TEST_HARNESS::kafkaBootstrapServers) | ||
| .withStaticServiceContext(TEST_HARNESS::getServiceContext) | ||
| .withProperty( | ||
| SERVER_PRECONDITIONS_CONFIG, | ||
| "io.confluent.ksql.rest.integration.PreconditionFunctionalTest$TestFailedPrecondition") | ||
| .buildWaitingOnPrecondition(latch); | ||
|
|
||
| @ClassRule | ||
| public static final RuleChain CHAIN = RuleChain | ||
| .outerRule(Retry.of(3, ZooKeeperClientException.class, 3, TimeUnit.SECONDS)) | ||
| .around(TEST_HARNESS) | ||
| .around(REST_APP); | ||
|
|
||
| @BeforeClass | ||
| public static void setUpClass() { | ||
| REST_APP.startAndWaitForPrecondition(); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldServeRequestsWhileWaitingForPrecondition() { | ||
| // When: | ||
| final KsqlErrorMessage error = | ||
| RestIntegrationTestUtil.makeKsqlRequestWithError(REST_APP, "SHOW STREAMS;"); | ||
|
|
||
| // Then: | ||
| assertThat(error.getErrorCode(), is(CUSTOM_ERROR_CODE)); | ||
| } | ||
|
|
||
| public static class TestFailedPrecondition implements KsqlServerPrecondition { | ||
|
|
||
| private boolean first = true; | ||
|
|
||
| @Override | ||
| public Optional<KsqlErrorMessage> checkPrecondition( | ||
| final KsqlRestConfig restConfig, | ||
| final ServiceContext serviceContext) { | ||
| return fail(); | ||
| } | ||
|
|
||
| private Optional<KsqlErrorMessage> fail() { | ||
| if (first) { | ||
| latch.countDown(); | ||
| first = false; | ||
| } | ||
| return Optional.of(new KsqlErrorMessage(CUSTOM_ERROR_CODE, "purposefully failed precondition")); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
77 changes: 77 additions & 0 deletions
77
...app/src/test/java/io/confluent/ksql/rest/server/TestKsqlRestAppWaitingOnPrecondition.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| /* | ||
| * Copyright 2020 Confluent Inc. | ||
| * | ||
| * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT | ||
| * WARRANTIES OF ANY KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations under the License. | ||
| */ | ||
|
|
||
| package io.confluent.ksql.rest.server; | ||
|
|
||
| import io.confluent.ksql.api.endpoints.KsqlSecurityContextProvider; | ||
| import io.confluent.ksql.rest.client.BasicCredentials; | ||
| import io.confluent.ksql.security.KsqlSecurityExtension; | ||
| import io.confluent.ksql.services.ServiceContext; | ||
| import io.confluent.ksql.util.KsqlConfig; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import java.util.concurrent.CountDownLatch; | ||
| import java.util.function.BiFunction; | ||
| import java.util.function.Supplier; | ||
| import org.glassfish.hk2.utilities.Binder; | ||
|
|
||
| /** | ||
| * A {@link TestKsqlRestApp} for testing behavior of a server stuck waiting on a precondition. | ||
| * | ||
| * The server is not started automatically. Rather, {@code startAndWaitForPrecondition()} should | ||
| * be called by the test suite. A {@code CountDownLatch} provided in the constructor counts down | ||
| * when a precondition check is called, in order to finish configuring the app once the server | ||
| * is waiting for preconditions. | ||
| */ | ||
| public class TestKsqlRestAppWaitingOnPrecondition extends TestKsqlRestApp { | ||
|
|
||
| private CountDownLatch latch; | ||
|
|
||
| TestKsqlRestAppWaitingOnPrecondition( | ||
| final Supplier<String> bootstrapServers, | ||
| final Map<String, Object> additionalProps, | ||
| final Supplier<ServiceContext> serviceContext, | ||
| final BiFunction<KsqlConfig, KsqlSecurityExtension, Binder> serviceContextBinderFactory, | ||
| final KsqlSecurityContextProvider securityContextProvider, | ||
| final Optional<BasicCredentials> credentials, | ||
| final CountDownLatch latch | ||
| ) { | ||
| super(bootstrapServers, additionalProps, serviceContext, serviceContextBinderFactory, securityContextProvider, credentials); | ||
| this.latch = latch; | ||
| } | ||
|
|
||
| @Override | ||
| protected void before() { | ||
| initialize(); | ||
| } | ||
|
|
||
| public void startAndWaitForPrecondition() { | ||
| try { | ||
| new Thread(() -> { | ||
| try { | ||
| restServer.startAsync(); | ||
| } catch (Exception e) { | ||
| throw new RuntimeException("Error starting server", e); | ||
| } | ||
| }).start(); | ||
| latch.await(); | ||
| } catch (final Exception var2) { | ||
| throw new RuntimeException("Failed to start Ksql rest server", var2); | ||
| } | ||
|
|
||
| listeners.addAll(ksqlRestApplication.getListeners()); | ||
| ksqlEngine = ksqlRestApplication.getEngine(); | ||
| } | ||
| } |