You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by ni...@apache.org on 2018/09/21 06:08:43 UTC
[incubator-servicecomb-saga] 04/05: SCB-909 Refactor retry logic
when grpc sending error.
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit 7f260e68d8b4e373ac7b207e96ad4c9c42100c30
Author: cherrylzhao <zh...@126.com>
AuthorDate: Thu Sep 20 16:16:01 2018 +0800
SCB-909 Refactor retry logic when grpc sending error.
---
.../omega/connector/grpc/GrpcTccEventService.java | 2 +-
.../connector/grpc/PushBackReconnectRunnable.java | 20 ++++-
.../grpc/tcc/GrpcCoordinateStreamObserver.java | 11 +--
.../connector/grpc/tcc/GrpcOnErrorHandler.java | 85 ++++++++++++++++++++++
.../grpc/tcc/GrpcTccClientMessageSender.java | 3 +-
.../connector/grpc/tcc/LoadBalanceContext.java | 56 +++-----------
.../grpc/tcc/LoadBalanceSenderAdapter.java | 4 +-
.../grpc/tcc/ObserverErrorTaskFactory.java | 39 ----------
.../connector/grpc/tcc/PendingTaskRunner.java | 22 ++++--
.../grpc/tcc/ReconnectStreamObserver.java | 13 +++-
.../grpc/tcc/LoadBalanceContextBuilderTest.java | 4 +-
11 files changed, 153 insertions(+), 106 deletions(-)
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccEventService.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccEventService.java
index c491d4b..c619584 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccEventService.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/GrpcTccEventService.java
@@ -55,7 +55,7 @@ public class GrpcTccEventService implements TccEventService {
tccBlockingEventService = TccEventServiceGrpc.newBlockingStub(channel);
tccAsyncEventService = TccEventServiceGrpc.newStub(channel);
this.serviceConfig = serviceConfig(serviceConfig.serviceName(), serviceConfig.instanceId());
- observer = new GrpcCoordinateStreamObserver(handler);
+ observer = new GrpcCoordinateStreamObserver(null, null, handler);
}
@Override
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnable.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnable.java
index 6bd0cc6..6faffcb 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnable.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/PushBackReconnectRunnable.java
@@ -19,8 +19,8 @@ package org.apache.servicecomb.saga.omega.connector.grpc;
import java.lang.invoke.MethodHandles;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.BlockingQueue;
-
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,6 +29,7 @@ public class PushBackReconnectRunnable implements Runnable {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private final MessageSender messageSender;
private final Map<MessageSender, Long> senders;
+
private final BlockingQueue<Runnable> pendingTasks;
private final BlockingQueue<MessageSender> connectedSenders;
@@ -58,4 +59,21 @@ public class PushBackReconnectRunnable implements Runnable {
pendingTasks.offer(this);
}
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PushBackReconnectRunnable that = (PushBackReconnectRunnable) o;
+ return Objects.equals(messageSender, that.messageSender);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(messageSender);
+ }
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcCoordinateStreamObserver.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcCoordinateStreamObserver.java
index 4f55058..d69930c 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcCoordinateStreamObserver.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcCoordinateStreamObserver.java
@@ -18,6 +18,7 @@
package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
import java.lang.invoke.MethodHandles;
+import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.apache.servicecomb.saga.omega.transaction.tcc.TccMessageHandler;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccCoordinateCommand;
import org.slf4j.Logger;
@@ -26,17 +27,13 @@ import org.slf4j.LoggerFactory;
public class GrpcCoordinateStreamObserver extends ReconnectStreamObserver<GrpcTccCoordinateCommand> {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final TccMessageHandler messageHandler;
- public GrpcCoordinateStreamObserver(Runnable errorHandler,
- TccMessageHandler messageHandler) {
- super(errorHandler);
- this.messageHandler = messageHandler;
- }
+ private final TccMessageHandler messageHandler;
public GrpcCoordinateStreamObserver(
+ LoadBalanceContext loadBalanceContext, MessageSender messageSender,
TccMessageHandler messageHandler) {
- super(null);
+ super(loadBalanceContext, messageSender);
this.messageHandler = messageHandler;
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcOnErrorHandler.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcOnErrorHandler.java
new file mode 100644
index 0000000..2f35896
--- /dev/null
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcOnErrorHandler.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (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.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
+
+import com.google.common.base.Supplier;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import org.apache.servicecomb.saga.omega.connector.grpc.PushBackReconnectRunnable;
+import org.apache.servicecomb.saga.omega.connector.grpc.RetryableMessageSender;
+import org.apache.servicecomb.saga.omega.transaction.MessageSender;
+
+public class GrpcOnErrorHandler {
+
+ private final BlockingQueue<Runnable> pendingTasks;
+
+ private final Map<MessageSender, Long> senders;
+
+ private final GrpcRetryContext grpcRetryContext;
+
+ public GrpcOnErrorHandler(BlockingQueue<Runnable> pendingTasks, Map<MessageSender, Long> senders) {
+ this.pendingTasks = pendingTasks;
+ this.senders = senders;
+ this.grpcRetryContext = new GrpcRetryContext();
+ }
+
+ public void handle(MessageSender messageSender) {
+ final Runnable runnable = new PushBackReconnectRunnable(
+ messageSender,
+ senders,
+ pendingTasks,
+ grpcRetryContext.getReconnectedSenders()
+ );
+ synchronized (pendingTasks) {
+ if (!pendingTasks.contains(runnable)) {
+ pendingTasks.offer(runnable);
+ }
+ }
+ }
+
+ public GrpcRetryContext getGrpcRetryContext() {
+ return grpcRetryContext;
+ }
+
+ public static class GrpcRetryContext {
+
+ private final BlockingQueue<MessageSender> reconnectedSenders = new LinkedBlockingQueue<>();
+
+ private final MessageSender retryMessageSender = new RetryableMessageSender(reconnectedSenders);
+
+ private final Supplier<MessageSender> defaultMessageSender = new Supplier<MessageSender>() {
+ @Override
+ public MessageSender get() {
+ return retryMessageSender;
+ }
+ };
+
+ public BlockingQueue<MessageSender> getReconnectedSenders() {
+ return reconnectedSenders;
+ }
+
+ public MessageSender getRetryMessageSender() {
+ return retryMessageSender;
+ }
+
+ public Supplier<MessageSender> getDefaultMessageSender() {
+ return defaultMessageSender;
+ }
+ }
+}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
index 9eb7b6a..e23f366 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/GrpcTccClientMessageSender.java
@@ -53,8 +53,7 @@ public class GrpcTccClientMessageSender implements TccMessageSender {
tccBlockingEventService = TccEventServiceGrpc.newBlockingStub(channel);
tccAsyncEventService = TccEventServiceGrpc.newStub(channel);
this.serviceConfig = serviceConfig(serviceConfig.serviceName(), serviceConfig.instanceId());
- Runnable errorTask = ObserverErrorTaskFactory.createTask(this, loadContext);
- observer = new GrpcCoordinateStreamObserver(errorTask, handler);
+ observer = new GrpcCoordinateStreamObserver(loadContext, this, handler);
}
@Override
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContext.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContext.java
index a2a8bac..d8ac080 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContext.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContext.java
@@ -17,13 +17,9 @@
package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
-import com.google.common.base.Supplier;
import io.grpc.ManagedChannel;
import java.util.Collection;
import java.util.Map;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import org.apache.servicecomb.saga.omega.connector.grpc.RetryableMessageSender;
import org.apache.servicecomb.saga.omega.transaction.MessageSender;
public class LoadBalanceContext {
@@ -32,63 +28,35 @@ public class LoadBalanceContext {
private final Collection<ManagedChannel> channels;
- private final int reconnectDelay;
+ private final PendingTaskRunner pendingTaskRunner;
- private final BlockingQueue<Runnable> pendingTasks = new LinkedBlockingQueue<>();
+ private final GrpcOnErrorHandler grpcOnErrorHandler;
- private final BlockingQueue<MessageSender> availableMessageSenders = new LinkedBlockingQueue<>();
-
- private final MessageSender retryableMessageSender = new RetryableMessageSender(availableMessageSenders);
-
- private final Supplier<MessageSender> defaultMessageSender = new Supplier<MessageSender>() {
- @Override
- public MessageSender get() {
- return retryableMessageSender;
- }
- };
-
- private final PendingTaskRunner pendingTaskRunner = new PendingTaskRunner(this);
-
- public LoadBalanceContext(
- Map<MessageSender, Long> senders, Collection<ManagedChannel> channels, int reconnectDelay) {
+ public LoadBalanceContext(Map<MessageSender, Long> senders,
+ Collection<ManagedChannel> channels, int reconnectDelay) {
this.senders = senders;
this.channels = channels;
- this.reconnectDelay = reconnectDelay;
+ this.pendingTaskRunner = new PendingTaskRunner(reconnectDelay);
+ this.grpcOnErrorHandler = new GrpcOnErrorHandler(pendingTaskRunner.getPendingTasks(), senders);
}
public Map<MessageSender, Long> getSenders() {
return senders;
}
- public void setSenders(Map<MessageSender, Long> senders) {
- this.senders = senders;
- }
-
public Collection<ManagedChannel> getChannels() {
return channels;
}
- public int getReconnectDelay() {
- return reconnectDelay;
- }
-
- public BlockingQueue<Runnable> getPendingTasks() {
- return pendingTasks;
- }
-
- public BlockingQueue<MessageSender> getAvailableMessageSenders() {
- return availableMessageSenders;
- }
-
- public MessageSender getRetryableMessageSender() {
- return retryableMessageSender;
- }
-
public PendingTaskRunner getPendingTaskRunner() {
return pendingTaskRunner;
}
- public Supplier<MessageSender> getDefaultMessageSender() {
- return defaultMessageSender;
+ public GrpcOnErrorHandler getGrpcOnErrorHandler() {
+ return grpcOnErrorHandler;
+ }
+
+ public void setSenders(Map<MessageSender, Long> senders) {
+ this.senders = senders;
}
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java
index 2b6f797..eb0dc08 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceSenderAdapter.java
@@ -45,7 +45,8 @@ public abstract class LoadBalanceSenderAdapter implements MessageSender {
@SuppressWarnings("unchecked")
public <T> T pickMessageSender() {
- return (T) senderPicker.pick(loadContext.getSenders(), loadContext.getDefaultMessageSender());
+ return (T) senderPicker.pick(loadContext.getSenders(),
+ loadContext.getGrpcOnErrorHandler().getGrpcRetryContext().getDefaultMessageSender());
}
public <T> Optional<AlphaResponse> doGrpcSend(MessageSender messageSender, T event, SenderExecutor<T> executor) {
@@ -59,6 +60,7 @@ public abstract class LoadBalanceSenderAdapter implements MessageSender {
} catch (Exception e) {
LOG.error("Retry sending event {} due to failure", event, e);
loadContext.getSenders().put(messageSender, Long.MAX_VALUE);
+ loadContext.getGrpcOnErrorHandler().handle(messageSender);
}
return Optional.fromNullable(response);
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/ObserverErrorTaskFactory.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/ObserverErrorTaskFactory.java
deleted file mode 100644
index ac1428f..0000000
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/ObserverErrorTaskFactory.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (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.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
-
-import org.apache.servicecomb.saga.omega.connector.grpc.PushBackReconnectRunnable;
-import org.apache.servicecomb.saga.omega.transaction.MessageSender;
-
-public class ObserverErrorTaskFactory {
-
- static Runnable createTask(MessageSender messageSender, final LoadBalanceContext context) {
- final Runnable runnable = new PushBackReconnectRunnable(
- messageSender,
- context.getSenders(),
- context.getPendingTasks(),
- context.getAvailableMessageSenders()
- );
- return new Runnable() {
- @Override
- public void run() {
- context.getPendingTasks().offer(runnable);
- }
- };
- }
-}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/PendingTaskRunner.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/PendingTaskRunner.java
index b6dd98b..dfbfbd8 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/PendingTaskRunner.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/PendingTaskRunner.java
@@ -19,17 +19,21 @@ package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
public class PendingTaskRunner {
private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
- private final LoadBalanceContext loadContext;
+ private final BlockingQueue<Runnable> pendingTasks = new LinkedBlockingQueue<>();
- public PendingTaskRunner(LoadBalanceContext loadContext) {
- this.loadContext = loadContext;
+ private final int reconnectDelay;
+
+ public PendingTaskRunner(int reconnectDelay) {
+ this.reconnectDelay = reconnectDelay;
}
public void start() {
@@ -37,15 +41,23 @@ public class PendingTaskRunner {
@Override
public void run() {
try {
- loadContext.getPendingTasks().take().run();
+ pendingTasks.take().run();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
- }, 0, loadContext.getReconnectDelay(), MILLISECONDS);
+ }, 0, reconnectDelay, MILLISECONDS);
}
public void shutdown() {
scheduler.shutdown();
}
+
+ public BlockingQueue<Runnable> getPendingTasks() {
+ return pendingTasks;
+ }
+
+ public int getReconnectDelay() {
+ return reconnectDelay;
+ }
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/ReconnectStreamObserver.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/ReconnectStreamObserver.java
index cbf0c0e..938656f 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/ReconnectStreamObserver.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/ReconnectStreamObserver.java
@@ -19,6 +19,7 @@ package org.apache.servicecomb.saga.omega.connector.grpc.tcc;
import io.grpc.stub.StreamObserver;
import java.lang.invoke.MethodHandles;
+import org.apache.servicecomb.saga.omega.transaction.MessageSender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -26,16 +27,20 @@ public abstract class ReconnectStreamObserver<T> implements StreamObserver<T> {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final Runnable errorTask;
+ private final LoadBalanceContext loadContext;
- protected ReconnectStreamObserver(Runnable errorTask) {
- this.errorTask = errorTask;
+ private final MessageSender messageSender;
+
+ public ReconnectStreamObserver(
+ LoadBalanceContext loadContext, MessageSender messageSender) {
+ this.loadContext = loadContext;
+ this.messageSender = messageSender;
}
@Override
public void onError(Throwable t) {
LOG.error("Failed to process grpc coordinate command.", t);
- errorTask.run();
+ loadContext.getGrpcOnErrorHandler().handle(messageSender);
}
@Override
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java
index 2128ac7..06a0851 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/saga/omega/connector/grpc/tcc/LoadBalanceContextBuilderTest.java
@@ -75,7 +75,7 @@ public class LoadBalanceContextBuilderTest {
when(clusterConfig.isEnableSSL()).thenReturn(false);
LoadBalanceContext loadContext = tccLoadBalanceContextBuilder.build();
- assertThat(loadContext.getReconnectDelay(), is(30));
+ assertThat(loadContext.getPendingTaskRunner().getReconnectDelay(), is(30));
assertThat(loadContext.getSenders().size(), is(2));
assertThat(loadContext.getSenders().keySet().iterator().next(), instanceOf(TccMessageSender.class));
assertThat(loadContext.getSenders().values().iterator().next(), is(0l));
@@ -91,7 +91,7 @@ public class LoadBalanceContextBuilderTest {
when(clusterConfig.getCertChain()).thenReturn(getClass().getClassLoader().getResource("ca.crt").getFile());
when(clusterConfig.getKey()).thenReturn(getClass().getClassLoader().getResource("client.pem").getFile());
LoadBalanceContext loadContext = tccLoadBalanceContextBuilder.build();
- assertThat(loadContext.getReconnectDelay(), is(30));
+ assertThat(loadContext.getPendingTaskRunner().getReconnectDelay(), is(30));
assertThat(loadContext.getSenders().size(), is(2));
assertThat(loadContext.getSenders().keySet().iterator().next(), instanceOf(TccMessageSender.class));
assertThat(loadContext.getSenders().values().iterator().next(), is(0l));