Skip to content

Commit

Permalink
Set function channel to idle to prevent DNS resolution of deleted pod (…
Browse files Browse the repository at this point in the history 
…apache#14750)
  • Loading branch information
@michaeljmarshall
michaeljmarshall authored Mar 21, 2022
1 parent c97e3f1 commit 1e619ce
Show file tree
Hide file tree
Showing 3 changed files with 26 additions and 17 deletions.
15 changes: 12 additions & 3 deletions ...ntime/src/main/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntime.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -336,6 +336,15 @@ public void join() throws Exception {

@Override
public void stop() throws Exception {
// Because we're about to delete the function pods, we don't want the channel to attempt loading DNS, which
// will cease to exist if the deletion succeeds. However, we don't want to shut down the channel until
// the StatefulSet and Service are actually deleted.
if (channel != null) {
for (ManagedChannel cn : channel) {
cn.enterIdle();
}
}

deleteStatefulSet();
deleteService();

Expand Down Expand Up @@ -637,7 +646,7 @@ public void deleteStatefulSet() throws InterruptedException {


Actions.Action waitForStatefulSetDeletion = Actions.Action.builder()
.actionName(String.format("Waiting for statefulset for function %s to complete deletion", fqfn))
.actionName(String.format("Waiting for StatefulSet deletion to complete deletion of function %s", fqfn))
// set retry period to be about 2x the graceshutdown time
.numRetries(KubernetesRuntimeFactory.numRetries * 2)
.sleepBetweenInvocationsMs(KubernetesRuntimeFactory.sleepBetweenRetriesMs * 2)
Expand Down Expand Up @@ -786,7 +795,7 @@ public void deleteService() throws InterruptedException {
.build();

Actions.Action waitForServiceDeletion = Actions.Action.builder()
.actionName(String.format("Waiting for statefulset for function %s to complete deletion", fqfn))
.actionName(String.format("Waiting for service deletion to complete deletion of function %s", fqfn))
.numRetries(KubernetesRuntimeFactory.numRetries)
.sleepBetweenInvocationsMs(KubernetesRuntimeFactory.sleepBetweenRetriesMs)
.supplier(() -> {
Expand All @@ -796,7 +805,7 @@ public void deleteService() throws InterruptedException {
null, null, null);

} catch (ApiException e) {
// statefulset is gone
// service is gone
if (e.getCode() == HTTP_NOT_FOUND) {
return Actions.ActionResult.builder().success(true).build();
}
Expand Down
2 changes: 1 addition & 1 deletion pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/Actions.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ private boolean runAction(Action action) throws InterruptedException {
ActionResult actionResult = action.getSupplier().get();

if (actionResult.isSuccess()) {
log.info("Sucessfully completed action [ {} ]", action.getActionName());
log.info("Successfully completed action [ {} ]", action.getActionName());
if (action.getOnSuccess() != null) {
action.getOnSuccess().accept(actionResult);
}
Expand Down
26 changes: 13 additions & 13 deletions pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/ActionsTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ public void testActionsSuccess() throws InterruptedException {
when(supplier2.get()).thenReturn(Actions.ActionResult.builder().success(true).build());

java.util.function.Consumer<Actions.ActionResult> onFail = mock(java.util.function.Consumer.class);
java.util.function.Consumer<Actions.ActionResult> onSucess = mock(java.util.function.Consumer.class);
java.util.function.Consumer<Actions.ActionResult> onSuccess = mock(java.util.function.Consumer.class);

Actions.Action action1 = spy(
Actions.Action.builder()
Expand All @@ -54,7 +54,7 @@ public void testActionsSuccess() throws InterruptedException {
.supplier(supplier1)
.continueOn(true)
.onFail(onFail)
.onSuccess(onSucess)
.onSuccess(onSuccess)
.build());

Actions.Action action2 = spy(
Expand All @@ -73,7 +73,7 @@ public void testActionsSuccess() throws InterruptedException {
assertEquals(actions.numActions(), 2);
verify(supplier1, times(1)).get();
verify(onFail, times(0)).accept(any());
verify(onSucess, times(1)).accept(any());
verify(onSuccess, times(1)).accept(any());
verify(supplier2, times(1)).get();
}

Expand All @@ -88,7 +88,7 @@ public void testActionsOneAction() throws InterruptedException {
when(supplier2.get()).thenReturn(Actions.ActionResult.builder().success(true).build());

java.util.function.Consumer<Actions.ActionResult> onFail = mock(java.util.function.Consumer.class);
java.util.function.Consumer<Actions.ActionResult> onSucess = mock(java.util.function.Consumer.class);
java.util.function.Consumer<Actions.ActionResult> onSuccess = mock(java.util.function.Consumer.class);

Actions.Action action1 = spy(
Actions.Action.builder()
Expand All @@ -98,7 +98,7 @@ public void testActionsOneAction() throws InterruptedException {
.supplier(supplier1)
.continueOn(false)
.onFail(onFail)
.onSuccess(onSucess)
.onSuccess(onSuccess)
.build());
Actions.Action action2 = spy(
Actions.Action.builder()
Expand All @@ -107,7 +107,7 @@ public void testActionsOneAction() throws InterruptedException {
.sleepBetweenInvocationsMs(200)
.supplier(supplier2)
.onFail(onFail)
.onSuccess(onSucess)
.onSuccess(onSuccess)
.build());

Actions actions = Actions.newBuilder()
Expand All @@ -118,7 +118,7 @@ public void testActionsOneAction() throws InterruptedException {
assertEquals(actions.numActions(), 2);
verify(supplier1, times(1)).get();
verify(onFail, times(0)).accept(any());
verify(onSucess, times(1)).accept(any());
verify(onSuccess, times(1)).accept(any());
verify(supplier2, times(0)).get();
}

Expand All @@ -134,7 +134,7 @@ public void testActionsRetry() throws InterruptedException {
when(supplier2.get()).thenReturn(Actions.ActionResult.builder().success(true).build());

java.util.function.Consumer<Actions.ActionResult> onFail = mock(java.util.function.Consumer.class);
java.util.function.Consumer<Actions.ActionResult> onSucess = mock(java.util.function.Consumer.class);
java.util.function.Consumer<Actions.ActionResult> onSuccess = mock(java.util.function.Consumer.class);

Actions.Action action1 = spy(
Actions.Action.builder()
Expand All @@ -144,7 +144,7 @@ public void testActionsRetry() throws InterruptedException {
.supplier(supplier1)
.continueOn(false)
.onFail(onFail)
.onSuccess(onSucess)
.onSuccess(onSuccess)
.build());

Actions.Action action2 = spy(
Expand All @@ -163,7 +163,7 @@ public void testActionsRetry() throws InterruptedException {
assertEquals(actions.numActions(), 2);
verify(supplier1, times(11)).get();
verify(onFail, times(1)).accept(any());
verify(onSucess, times(0)).accept(any());
verify(onSuccess, times(0)).accept(any());
verify(supplier2, times(1)).get();
}

Expand All @@ -177,7 +177,7 @@ public void testActionsNoContinueOn() throws InterruptedException {
when(supplier2.get()).thenReturn(Actions.ActionResult.builder().success(true).build());

java.util.function.Consumer<Actions.ActionResult> onFail = mock(java.util.function.Consumer.class);
java.util.function.Consumer<Actions.ActionResult> onSucess = mock(java.util.function.Consumer.class);
java.util.function.Consumer<Actions.ActionResult> onSuccess = mock(java.util.function.Consumer.class);

Actions.Action action1 = spy(
Actions.Action.builder()
Expand All @@ -186,7 +186,7 @@ public void testActionsNoContinueOn() throws InterruptedException {
.sleepBetweenInvocationsMs(100)
.supplier(supplier1)
.onFail(onFail)
.onSuccess(onSucess)
.onSuccess(onSuccess)
.build());

Actions.Action action2 = spy(
Expand All @@ -205,7 +205,7 @@ public void testActionsNoContinueOn() throws InterruptedException {
assertEquals(actions.numActions(), 2);
verify(supplier1, times(1)).get();
verify(onFail, times(0)).accept(any());
verify(onSucess, times(1)).accept(any());
verify(onSuccess, times(1)).accept(any());
verify(supplier2, times(1)).get();
}
}

0 comments on commit 1e619ce

Please sign in to comment.