#194

Committed 10 Oct 2023 03:10PM UTC coverage: 77.401% (+0.005%) from 77.396%

Build # #194

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Reset lastHandledEventId on speculative WFT (#1881)

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

18697 of 24156 relevant lines covered (77.4%)

0.77 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.4

/temporal-sdk/src/main/java/io/temporal/internal/worker/Poller.java

/*
 * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
 *
 * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Modifications copyright (C) 2017 Uber Technologies, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this material 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 io.temporal.internal.worker;

import com.uber.m3.tally.Scope;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.temporal.internal.BackoffThrottler;
import io.temporal.internal.common.GrpcUtils;
import io.temporal.worker.MetricsType;
import java.time.Duration;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

final class Poller<T> implements SuspendableWorker {

  public interface PollTask<TT> {
    /**
     * Pollers should shade or wrap all {@code java.lang.InterruptedException}s and raise {@code
     * Thread.interrupted()} flag. This follows GRPC stubs approach, see {@code
     * io.grpc.stub.ClientCalls#blockingUnaryCall}. Because pollers use GRPC stubs anyway, we chose
     * this implementation for consistency. The caller of the poll task is responsible for handling
     * the flag.
     *
     * @return result of the task
     */
    TT poll();
  }

  interface ThrowingRunnable {
    void run() throws Throwable;
  }

  private final String identity;
  private final ShutdownableTaskExecutor<T> taskExecutor;
  private final PollTask<T> pollTask;
  private final PollerOptions pollerOptions;
  private static final Logger log = LoggerFactory.getLogger(Poller.class);
  private ThreadPoolExecutor pollExecutor;
  private final Scope workerMetricsScope;

  private final AtomicReference<CountDownLatch> suspendLatch = new AtomicReference<>();

  private Throttler pollRateThrottler;

  private final Thread.UncaughtExceptionHandler uncaughtExceptionHandler =
      new PollerUncaughtExceptionHandler();

  public Poller(
      String identity,
      PollTask<T> pollTask,
      ShutdownableTaskExecutor<T> taskExecutor,
      PollerOptions pollerOptions,
      Scope workerMetricsScope) {
    Objects.requireNonNull(identity, "identity cannot be null");
    Objects.requireNonNull(pollTask, "poll service should not be null");
    Objects.requireNonNull(taskExecutor, "taskExecutor should not be null");
    Objects.requireNonNull(pollerOptions, "pollerOptions should not be null");
    Objects.requireNonNull(workerMetricsScope, "workerMetricsScope should not be null");

    this.identity = identity;
    this.pollTask = pollTask;
    this.taskExecutor = taskExecutor;
    this.pollerOptions = pollerOptions;
    this.workerMetricsScope = workerMetricsScope;
  }

  @Override
  public boolean start() {
    log.info("start: {}", this);

    if (pollerOptions.getMaximumPollRatePerSecond() > 0.0) {
      pollRateThrottler =
          new Throttler(
              "poller",
              pollerOptions.getMaximumPollRatePerSecond(),
              pollerOptions.getMaximumPollRateIntervalMilliseconds());
    }

    // It is important to pass blocking queue of at least options.getPollThreadCount() capacity. As
    // task enqueues next task the buffering is needed to queue task until the previous one releases
    // a thread.
    pollExecutor =
        new ThreadPoolExecutor(
            pollerOptions.getPollThreadCount(),
            pollerOptions.getPollThreadCount(),
            1,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(pollerOptions.getPollThreadCount()));
    pollExecutor.setThreadFactory(
        new ExecutorThreadFactory(
            pollerOptions.getPollThreadNamePrefix(), pollerOptions.getUncaughtExceptionHandler()));

    for (int i = 0; i < pollerOptions.getPollThreadCount(); i++) {
      pollExecutor.execute(new PollLoopTask(new PollExecutionTask()));
      workerMetricsScope.counter(MetricsType.POLLER_START_COUNTER).inc(1);
    }

    return true;
  }

  @Override
  public CompletableFuture<Void> shutdown(ShutdownManager shutdownManager, boolean interruptTasks) {
    log.info("shutdown: {}", this);
    WorkerLifecycleState lifecycleState = getLifecycleState();
    switch (lifecycleState) {
      case NOT_STARTED:
      case TERMINATED:
        return CompletableFuture.completedFuture(null);
    }

    return shutdownManager
        // it's ok to forcefully shutdown pollers, especially because they stuck in a long poll call
        // we don't lose any progress doing that
        .shutdownExecutorNow(pollExecutor, this + "#pollExecutor", Duration.ofSeconds(1))
        .exceptionally(
            e -> {
              log.error("Unexpected exception during shutdown", e);
              return null;
            });
  }

  @Override
  public void awaitTermination(long timeout, TimeUnit unit) {
    WorkerLifecycleState lifecycleState = getLifecycleState();
    switch (lifecycleState) {
      case NOT_STARTED:
      case TERMINATED:
        return;
    }

    long timeoutMillis = unit.toMillis(timeout);
    ShutdownManager.awaitTermination(pollExecutor, timeoutMillis);
  }

  @Override
  public void suspendPolling() {
    if (suspendLatch.compareAndSet(null, new CountDownLatch(1))) {
      log.info("Suspend Polling: {}", this);
    } else {
      log.info("Polling is already suspended: {}", this);
    }
  }

  @Override
  public void resumePolling() {
    CountDownLatch existing = suspendLatch.getAndSet(null);
    if (existing != null) {
      log.info("Resume Polling {}", this);
      existing.countDown();
    }
  }

  @Override
  public boolean isSuspended() {
    return suspendLatch.get() != null;
  }

  @Override
  public boolean isShutdown() {
    return pollExecutor.isShutdown();
  }

  @Override
  public boolean isTerminated() {
    return pollExecutor.isTerminated() && taskExecutor.isTerminated();
  }

  @Override
  public WorkerLifecycleState getLifecycleState() {
    if (pollExecutor == null) {
      return WorkerLifecycleState.NOT_STARTED;
    }
    if (suspendLatch.get() != null) {
      return WorkerLifecycleState.SUSPENDED;
    }
    if (pollExecutor.isShutdown()) {
      // return TERMINATED only if both pollExecutor and taskExecutor are terminated
      if (pollExecutor.isTerminated() && taskExecutor.isTerminated()) {
        return WorkerLifecycleState.TERMINATED;
      } else {
        return WorkerLifecycleState.SHUTDOWN;
      }
    }
    return WorkerLifecycleState.ACTIVE;
  }

  @Override
  public String toString() {
    // TODO using pollThreadNamePrefix here is ugly. We should consider introducing some concept of
    // WorkerContext [workerIdentity, namespace, queue, local/non-local if applicable] and pass it
    // around
    // that will simplify such kind of logging through workers.
    return String.format(
        "Poller{name=%s, identity=%s}", pollerOptions.getPollThreadNamePrefix(), identity);
  }

  private class PollLoopTask implements Runnable {

    private final Poller.ThrowingRunnable task;
    private final BackoffThrottler pollBackoffThrottler;

    PollLoopTask(Poller.ThrowingRunnable task) {
      this.task = task;
      this.pollBackoffThrottler =
          new BackoffThrottler(
              pollerOptions.getBackoffInitialInterval(),
              pollerOptions.getBackoffCongestionInitialInterval(),
              pollerOptions.getBackoffMaximumInterval(),
              pollerOptions.getBackoffCoefficient(),
              pollerOptions.getBackoffMaximumJitterCoefficient());
    }

    @Override
    public void run() {
      try {
        long throttleMs = pollBackoffThrottler.getSleepTime();
        if (throttleMs > 0) {
          Thread.sleep(throttleMs);
        }
        if (pollRateThrottler != null) {
          pollRateThrottler.throttle();
        }

        CountDownLatch suspender = Poller.this.suspendLatch.get();
        if (suspender != null) {
          if (log.isDebugEnabled()) {
            log.debug("poll task suspending latchCount=" + suspender.getCount());
          }
          suspender.await();
        }

        if (shouldTerminate()) {
          return;
        }

        task.run();
        pollBackoffThrottler.success();
      } catch (Throwable e) {
        if (e instanceof InterruptedException) {
          // we restore the flag here, so it can be checked and processed (with exit) in finally.
          Thread.currentThread().interrupt();
        } else {
          // Don't increase throttle on InterruptedException
          pollBackoffThrottler.failure(
              (e instanceof StatusRuntimeException)
                  ? ((StatusRuntimeException) e).getStatus().getCode()
                  : Status.Code.UNKNOWN);
        }
        uncaughtExceptionHandler.uncaughtException(Thread.currentThread(), e);
      } finally {
        if (!shouldTerminate()) {
          // Resubmit itself back to pollExecutor
          pollExecutor.execute(this);
        } else {
          log.info("poll loop is terminated: {}", Poller.this.pollTask.getClass().getSimpleName());
        }
      }
    }

    /**
     * Defines if the task should be terminated.
     *
     * <p>This method preserves the interrupted flag of the current thread.
     *
     * @return true if pollExecutor is terminating, or the current thread is interrupted.
     */
    private boolean shouldTerminate() {
      return pollExecutor.isShutdown() || Thread.currentThread().isInterrupted();
    }
  }

  private class PollExecutionTask implements Poller.ThrowingRunnable {

    @Override
    public void run() throws Exception {
      T task = pollTask.poll();
      if (task != null) {
        taskExecutor.process(task);
      }
    }
  }

  private final class PollerUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {

    @Override
    public void uncaughtException(Thread t, Throwable e) {
      if (!pollExecutor.isShutdown() || !shouldIgnoreDuringShutdown(e)) {
        logPollErrors(t, e);
      } else {
        logPollExceptionsSuppressedDuringShutdown(t, e);
      }
    }

    private void logPollErrors(Thread t, Throwable e) {
      if (e instanceof StatusRuntimeException) {
        StatusRuntimeException te = (StatusRuntimeException) e;
        if (te.getStatus().getCode() == Status.Code.DEADLINE_EXCEEDED) {
          log.info("DEADLINE_EXCEEDED in poller thread {}", t.getName(), e);
          return;
        }
      }
      log.warn("Failure in poller thread {}", t.getName(), e);
    }

    /**
     * Some exceptions are considered normal during shutdown {@link #shouldIgnoreDuringShutdown} and
     * we log them in the most quite manner.
     *
     * @param t thread where the exception happened
     * @param e the exception itself
     */
    private void logPollExceptionsSuppressedDuringShutdown(Thread t, Throwable e) {
      log.trace(
          "Failure in thread {} is suppressed, considered normal during shutdown", t.getName(), e);
    }

    private boolean shouldIgnoreDuringShutdown(Throwable ex) {
      if (ex instanceof StatusRuntimeException) {
        if (GrpcUtils.isChannelShutdownException((StatusRuntimeException) ex)) {
          return true;
        }
      }
      return
      // if we are terminating and getting rejected execution - it's normal
      ex instanceof RejectedExecutionException
          // if the worker thread gets InterruptedException - it's normal during shutdown
          || ex instanceof InterruptedException
          // if we get wrapped InterruptedException like what PollTask or GRPC clients do with
          // setting Thread.interrupted() on - it's normal during shutdown too. See PollTask
          // javadoc.
          || ex.getCause() instanceof InterruptedException;
    }
  }
}

1	/*
2	* Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
3	*
4	* Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
5	*
6	* Modifications copyright (C) 2017 Uber Technologies, Inc.
7	*
8	* Licensed under the Apache License, Version 2.0 (the "License");
9	* you may not use this material except in compliance with the License.
10	* You may obtain a copy of the License at
11	*
12	* http://www.apache.org/licenses/LICENSE-2.0
13	*
14	* Unless required by applicable law or agreed to in writing, software
15	* distributed under the License is distributed on an "AS IS" BASIS,
16	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17	* See the License for the specific language governing permissions and
18	* limitations under the License.
19	*/
20
21	package io.temporal.internal.worker;
22
23	import com.uber.m3.tally.Scope;
24	import io.grpc.Status;
25	import io.grpc.StatusRuntimeException;
26	import io.temporal.internal.BackoffThrottler;
27	import io.temporal.internal.common.GrpcUtils;
28	import io.temporal.worker.MetricsType;
29	import java.time.Duration;
30	import java.util.Objects;
31	import java.util.concurrent.*;
32	import java.util.concurrent.atomic.AtomicReference;
33	import org.slf4j.Logger;
34	import org.slf4j.LoggerFactory;
35
36	final class Poller<T> implements SuspendableWorker {
37
38	public interface PollTask<TT> {
39	/**
40	* Pollers should shade or wrap all {@code java.lang.InterruptedException}s and raise {@code
41	* Thread.interrupted()} flag. This follows GRPC stubs approach, see {@code
42	* io.grpc.stub.ClientCalls#blockingUnaryCall}. Because pollers use GRPC stubs anyway, we chose
43	* this implementation for consistency. The caller of the poll task is responsible for handling
44	* the flag.
45	*
46	* @return result of the task
47	*/
48	TT poll();
49	}
50
51	interface ThrowingRunnable {
52	void run() throws Throwable;
53	}
54
55	private final String identity;
56	private final ShutdownableTaskExecutor<T> taskExecutor;
57	private final PollTask<T> pollTask;
58	private final PollerOptions pollerOptions;
59	private static final Logger log = LoggerFactory.getLogger(Poller.class);	1✔
60	private ThreadPoolExecutor pollExecutor;
61	private final Scope workerMetricsScope;
62
63	private final AtomicReference<CountDownLatch> suspendLatch = new AtomicReference<>();	1✔
64
65	private Throttler pollRateThrottler;
66
67	private final Thread.UncaughtExceptionHandler uncaughtExceptionHandler =	1✔
68	new PollerUncaughtExceptionHandler();
69
70	public Poller(
71	String identity,
72	PollTask<T> pollTask,
73	ShutdownableTaskExecutor<T> taskExecutor,
74	PollerOptions pollerOptions,
75	Scope workerMetricsScope) {	1✔
76	Objects.requireNonNull(identity, "identity cannot be null");	1✔
77	Objects.requireNonNull(pollTask, "poll service should not be null");	1✔
78	Objects.requireNonNull(taskExecutor, "taskExecutor should not be null");	1✔
79	Objects.requireNonNull(pollerOptions, "pollerOptions should not be null");	1✔
80	Objects.requireNonNull(workerMetricsScope, "workerMetricsScope should not be null");	1✔
81
82	this.identity = identity;	1✔
83	this.pollTask = pollTask;	1✔
84	this.taskExecutor = taskExecutor;	1✔
85	this.pollerOptions = pollerOptions;	1✔
86	this.workerMetricsScope = workerMetricsScope;	1✔
87	}	1✔
88
89	@Override
90	public boolean start() {
91	log.info("start: {}", this);	1✔
92
93	if (pollerOptions.getMaximumPollRatePerSecond() > 0.0) {	1✔
94	pollRateThrottler =	×
95	new Throttler(
96	"poller",
97	pollerOptions.getMaximumPollRatePerSecond(),	×
98	pollerOptions.getMaximumPollRateIntervalMilliseconds());	×
99	}
100
101	// It is important to pass blocking queue of at least options.getPollThreadCount() capacity. As
102	// task enqueues next task the buffering is needed to queue task until the previous one releases
103	// a thread.
104	pollExecutor =	1✔
105	new ThreadPoolExecutor(
106	pollerOptions.getPollThreadCount(),	1✔
107	pollerOptions.getPollThreadCount(),	1✔
108	1,
109	TimeUnit.SECONDS,
110	new ArrayBlockingQueue<>(pollerOptions.getPollThreadCount()));	1✔
111	pollExecutor.setThreadFactory(	1✔
112	new ExecutorThreadFactory(
113	pollerOptions.getPollThreadNamePrefix(), pollerOptions.getUncaughtExceptionHandler()));	1✔
114
115	for (int i = 0; i < pollerOptions.getPollThreadCount(); i++) {	1✔
116	pollExecutor.execute(new PollLoopTask(new PollExecutionTask()));	1✔
117	workerMetricsScope.counter(MetricsType.POLLER_START_COUNTER).inc(1);	1✔
118	}
119
120	return true;	1✔
121	}
122
123	@Override
124	public CompletableFuture<Void> shutdown(ShutdownManager shutdownManager, boolean interruptTasks) {
125	log.info("shutdown: {}", this);	1✔
126	WorkerLifecycleState lifecycleState = getLifecycleState();	1✔
127	switch (lifecycleState) {	1✔
128	case NOT_STARTED:
129	case TERMINATED:
130	return CompletableFuture.completedFuture(null);	1✔
131	}
132
133	return shutdownManager	1✔
134	// it's ok to forcefully shutdown pollers, especially because they stuck in a long poll call
135	// we don't lose any progress doing that
136	.shutdownExecutorNow(pollExecutor, this + "#pollExecutor", Duration.ofSeconds(1))	1✔
137	.exceptionally(	1✔
138	e -> {
139	log.error("Unexpected exception during shutdown", e);	×
140	return null;	×
141	});
142	}
143
144	@Override
145	public void awaitTermination(long timeout, TimeUnit unit) {
146	WorkerLifecycleState lifecycleState = getLifecycleState();	1✔
147	switch (lifecycleState) {	1✔
148	case NOT_STARTED:
149	case TERMINATED:
150	return;	1✔
151	}
152
153	long timeoutMillis = unit.toMillis(timeout);	1✔
154	ShutdownManager.awaitTermination(pollExecutor, timeoutMillis);	1✔
155	}	1✔
156
157	@Override
158	public void suspendPolling() {
159	if (suspendLatch.compareAndSet(null, new CountDownLatch(1))) {	1✔
160	log.info("Suspend Polling: {}", this);	1✔
161	} else {
162	log.info("Polling is already suspended: {}", this);	×
163	}
164	}	1✔
165
166	@Override
167	public void resumePolling() {
168	CountDownLatch existing = suspendLatch.getAndSet(null);	1✔
169	if (existing != null) {	1✔
170	log.info("Resume Polling {}", this);	1✔
171	existing.countDown();	1✔
172	}
173	}	1✔
174
175	@Override
176	public boolean isSuspended() {
177	return suspendLatch.get() != null;	1✔
178	}
179
180	@Override
181	public boolean isShutdown() {
182	return pollExecutor.isShutdown();	×
183	}
184
185	@Override
186	public boolean isTerminated() {
187	return pollExecutor.isTerminated() && taskExecutor.isTerminated();	×
188	}
189
190	@Override
191	public WorkerLifecycleState getLifecycleState() {
192	if (pollExecutor == null) {	1✔
193	return WorkerLifecycleState.NOT_STARTED;	×
194	}
195	if (suspendLatch.get() != null) {	1✔
196	return WorkerLifecycleState.SUSPENDED;	1✔
197	}
198	if (pollExecutor.isShutdown()) {	1✔
199	// return TERMINATED only if both pollExecutor and taskExecutor are terminated
200	if (pollExecutor.isTerminated() && taskExecutor.isTerminated()) {	1✔
201	return WorkerLifecycleState.TERMINATED;	1✔
202	} else {
203	return WorkerLifecycleState.SHUTDOWN;	1✔
204	}
205	}
206	return WorkerLifecycleState.ACTIVE;	1✔
207	}
208
209	@Override
210	public String toString() {
211	// TODO using pollThreadNamePrefix here is ugly. We should consider introducing some concept of
212	// WorkerContext [workerIdentity, namespace, queue, local/non-local if applicable] and pass it
213	// around
214	// that will simplify such kind of logging through workers.
215	return String.format(	1✔
216	"Poller{name=%s, identity=%s}", pollerOptions.getPollThreadNamePrefix(), identity);	1✔
217	}
218
219	private class PollLoopTask implements Runnable {
220
221	private final Poller.ThrowingRunnable task;
222	private final BackoffThrottler pollBackoffThrottler;
223
224	PollLoopTask(Poller.ThrowingRunnable task) {	1✔
225	this.task = task;	1✔
226	this.pollBackoffThrottler =	1✔
227	new BackoffThrottler(
228	pollerOptions.getBackoffInitialInterval(),	1✔
229	pollerOptions.getBackoffCongestionInitialInterval(),	1✔
230	pollerOptions.getBackoffMaximumInterval(),	1✔
231	pollerOptions.getBackoffCoefficient(),	1✔
232	pollerOptions.getBackoffMaximumJitterCoefficient());	1✔
233	}	1✔
234
235	@Override
236	public void run() {
237	try {
238	long throttleMs = pollBackoffThrottler.getSleepTime();	1✔
239	if (throttleMs > 0) {	1✔
240	Thread.sleep(throttleMs);	×
241	}
242	if (pollRateThrottler != null) {	1✔
243	pollRateThrottler.throttle();	×
244	}
245
246	CountDownLatch suspender = Poller.this.suspendLatch.get();	1✔
247	if (suspender != null) {	1✔
248	if (log.isDebugEnabled()) {	1✔
249	log.debug("poll task suspending latchCount=" + suspender.getCount());	×
250	}
251	suspender.await();	×
252	}
253
254	if (shouldTerminate()) {	1✔
255	return;	×
256	}
257
258	task.run();	1✔
259	pollBackoffThrottler.success();	1✔
260	} catch (Throwable e) {	1✔
261	if (e instanceof InterruptedException) {	1✔
262	// we restore the flag here, so it can be checked and processed (with exit) in finally.
263	Thread.currentThread().interrupt();	1✔
264	} else {
265	// Don't increase throttle on InterruptedException
266	pollBackoffThrottler.failure(	1✔
267	(e instanceof StatusRuntimeException)	1✔
268	? ((StatusRuntimeException) e).getStatus().getCode()	1✔
269	: Status.Code.UNKNOWN);	×
270	}
271	uncaughtExceptionHandler.uncaughtException(Thread.currentThread(), e);	1✔
272	} finally {
273	if (!shouldTerminate()) {	1✔
274	// Resubmit itself back to pollExecutor
275	pollExecutor.execute(this);	1✔
276	} else {
277	log.info("poll loop is terminated: {}", Poller.this.pollTask.getClass().getSimpleName());	1✔
278	}
279	}
280	}	1✔
281
282	/**
283	* Defines if the task should be terminated.
284	*
285	* <p>This method preserves the interrupted flag of the current thread.
286	*
287	* @return true if pollExecutor is terminating, or the current thread is interrupted.
288	*/
289	private boolean shouldTerminate() {
290	return pollExecutor.isShutdown() \|\| Thread.currentThread().isInterrupted();	1✔
291	}
292	}
293
294	private class PollExecutionTask implements Poller.ThrowingRunnable {	1✔
295
296	@Override
297	public void run() throws Exception {
298	T task = pollTask.poll();	1✔
299	if (task != null) {	1✔
300	taskExecutor.process(task);	1✔
301	}
302	}	1✔
303	}
304
305	private final class PollerUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {	1✔
306
307	@Override
308	public void uncaughtException(Thread t, Throwable e) {
309	if (!pollExecutor.isShutdown() \|\| !shouldIgnoreDuringShutdown(e)) {	1✔
310	logPollErrors(t, e);	×
311	} else {
312	logPollExceptionsSuppressedDuringShutdown(t, e);	1✔
313	}
314	}	1✔
315
316	private void logPollErrors(Thread t, Throwable e) {
317	if (e instanceof StatusRuntimeException) {	×
318	StatusRuntimeException te = (StatusRuntimeException) e;	×
319	if (te.getStatus().getCode() == Status.Code.DEADLINE_EXCEEDED) {	×
320	log.info("DEADLINE_EXCEEDED in poller thread {}", t.getName(), e);	×
321	return;	×
322	}
323	}
324	log.warn("Failure in poller thread {}", t.getName(), e);	×
325	}	×
326
327	/**
328	* Some exceptions are considered normal during shutdown {@link #shouldIgnoreDuringShutdown} and
329	* we log them in the most quite manner.
330	*
331	* @param t thread where the exception happened
332	* @param e the exception itself
333	*/
334	private void logPollExceptionsSuppressedDuringShutdown(Thread t, Throwable e) {
335	log.trace(	1✔
336	"Failure in thread {} is suppressed, considered normal during shutdown", t.getName(), e);	1✔
337	}	1✔
338
339	private boolean shouldIgnoreDuringShutdown(Throwable ex) {
340	if (ex instanceof StatusRuntimeException) {	1✔
341	if (GrpcUtils.isChannelShutdownException((StatusRuntimeException) ex)) {	1✔
342	return true;	×
343	}
344	}
345	return	1✔
346	// if we are terminating and getting rejected execution - it's normal
347	ex instanceof RejectedExecutionException
348	// if the worker thread gets InterruptedException - it's normal during shutdown
349	\|\| ex instanceof InterruptedException
350	// if we get wrapped InterruptedException like what PollTask or GRPC clients do with
351	// setting Thread.interrupted() on - it's normal during shutdown too. See PollTask
352	// javadoc.
353	\|\| ex.getCause() instanceof InterruptedException;	1✔
354	}
355	}
356	}

temporalio / sdk-java / #194

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous