public Object evaluate(
      final GenericRow row,
      final Object defaultValue,
      final ProcessingLogger logger,
      final Supplier<String> errorMsg
  ) {
    try {
      return expressionEvaluator.evaluate(new Object[]{
          spec.resolveArguments(row),
          defaultValue,
          logger,
          row
      });
    } catch (final Exception e) {
      final Throwable cause = e instanceof InvocationTargetException
          ? e.getCause()
          : e;

      logger.error(RecordProcessingError.recordProcessingError(errorMsg.get(), cause, row));
      return defaultValue;
    }
  }
@Test
  public void shouldPerformThreadSafeParameterEvaluation() throws Exception {
    // Given:
    spec.addParameter(
        ColumnName.of("foo1"),
        Integer.class,
        0
    );
    spec.addParameter(
        ColumnName.of("foo2"),
        Integer.class,
        1
    );

    final CountDownLatch threadLatch = new CountDownLatch(1);
    final CountDownLatch mainLatch = new CountDownLatch(1);

    final Map<String, Object> arguments1 = new HashMap<String, Object>() {{
      put("var0", 123);
      put("var1", 456);
    }};

    final Map<String, Object> arguments2 = new HashMap<String, Object>() {{
      put("var0", 100);
      put("var1", 200);
    }};

    when(expressionEvaluator.evaluate(new Object[]{arguments1, DEFAULT_VAL, processingLogger, genericRow(123, 456)}))
        .thenAnswer(
            invocation -> {
              threadLatch.countDown();
              assertThat(mainLatch.await(10, TimeUnit.SECONDS), is(true));
              return RETURN_VALUE;
            });

    compiledExpression = new CompiledExpression(
        expressionEvaluator,
        spec.build(),
        EXPRESSION_TYPE,
        expression
    );

    final Thread thread = new Thread(
        () -> compiledExpression
            .evaluate(genericRow(123, 456), DEFAULT_VAL, processingLogger, errorMsgSupplier)
    );

    // When:
    thread.start();

    // Then:
    assertThat(threadLatch.await(10, TimeUnit.SECONDS), is(true));

    // When:
    compiledExpression
        .evaluate(genericRow(100, 200), DEFAULT_VAL, processingLogger, errorMsgSupplier);
    mainLatch.countDown();

    // Then:
    thread.join();
    verify(expressionEvaluator, times(1))
        .evaluate(new Object[]{arguments1, DEFAULT_VAL, processingLogger, genericRow(123, 456)});
    verify(expressionEvaluator, times(1))
        .evaluate(new Object[]{arguments2, DEFAULT_VAL, processingLogger, genericRow(100, 200)});
  }