Skip to content

Commit

Permalink
Handle exceptions in waitForTemporalServerandLog (#3460)
Browse files Browse the repository at this point in the history
* Handle exceptions in waitForTemporalServerandLog

* Run gradlew format

Co-authored-by: Abhi Vaidyanatha <abhivaidyanatha@Abhis-MacBook-Pro.local>
  • Loading branch information
avaidyanatha and Abhi Vaidyanatha committed May 18, 2021
1 parent 588553e commit d4bcb18
Show file tree
Hide file tree
Showing 2 changed files with 72 additions and 3 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -84,12 +84,21 @@ public void run() {
factory.start();
}

private void waitForTemporalServerAndLog() {
protected void waitForTemporalServerAndLog() {
LOGGER.info("Waiting for temporal server...");

while (!getNamespaces(temporalService).contains("default")) {
boolean temporalStatus = false;

while (!temporalStatus) {
LOGGER.warn("Waiting for default namespace to be initialized in temporal...");
wait(2);

try {
temporalStatus = getNamespaces(temporalService).contains("default");
} catch (Exception e) {
// Ignore the exception because this likely means that the Temporal service is still initializing.
LOGGER.warn("Ignoring exception while trying to request Temporal namespaces:", e);
}
}

// sometimes it takes a few additional seconds for workflow queue listening to be available
Expand All @@ -106,7 +115,7 @@ private static void wait(int seconds) {
}
}

private static Set<String> getNamespaces(WorkflowServiceStubs temporalService) {
protected static Set<String> getNamespaces(WorkflowServiceStubs temporalService) {
return temporalService.blockingStub()
.listNamespaces(ListNamespacesRequest.newBuilder().build())
.getNamespacesList()
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
/*
* MIT License
*
* Copyright (c) 2020 Airbyte
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/

package io.airbyte.workers.temporal;

import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import io.airbyte.workers.process.ProcessBuilderFactory;
import io.temporal.api.namespace.v1.NamespaceInfo;
import io.temporal.api.workflowservice.v1.DescribeNamespaceResponse;
import io.temporal.serviceclient.WorkflowServiceStubs;
import java.nio.file.Path;
import java.util.List;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;

class TemporalPoolTest {

@Test
public void testWaitForTemporalServerOnLogThrowsException() {
WorkflowServiceStubs workflowServiceStubs = mock(WorkflowServiceStubs.class, Mockito.RETURNS_DEEP_STUBS);
Path path = mock(Path.class);
ProcessBuilderFactory processBuilderFactory = mock(ProcessBuilderFactory.class);
TemporalPool testPool = new TemporalPool(workflowServiceStubs, path, processBuilderFactory);
DescribeNamespaceResponse describeNamespaceResponse = mock(DescribeNamespaceResponse.class);
NamespaceInfo namespaceInfo = mock(NamespaceInfo.class);

when(namespaceInfo.getName()).thenReturn("default");
when(describeNamespaceResponse.getNamespaceInfo()).thenReturn(namespaceInfo);
when(workflowServiceStubs.blockingStub().listNamespaces(any()).getNamespacesList())
.thenThrow(RuntimeException.class)
.thenReturn(List.of(describeNamespaceResponse));
testPool.waitForTemporalServerAndLog();
}

}

0 comments on commit d4bcb18

Please sign in to comment.