public static <T> Write<T> write() {
    return new Write<>();
  }
@Test
  public void testWriteWithBackoff() throws Exception {
    String tableName = DatabaseTestHelper.getTestTableName("UT_WRITE_BACKOFF");
    DatabaseTestHelper.createTable(DATA_SOURCE, tableName);

    // lock table
    final Connection connection = DATA_SOURCE.getConnection();
    Statement lockStatement = connection.createStatement();
    lockStatement.execute("ALTER TABLE " + tableName + " LOCKSIZE TABLE");
    lockStatement.execute("LOCK TABLE " + tableName + " IN EXCLUSIVE MODE");

    // start a first transaction
    connection.setAutoCommit(false);
    PreparedStatement insertStatement =
        connection.prepareStatement("insert into " + tableName + " values(?, ?)");
    insertStatement.setInt(1, 1);
    insertStatement.setString(2, "TEST");
    insertStatement.execute();

    // try to write to this table
    pipeline
        .apply(Create.of(Collections.singletonList(KV.of(1, "TEST"))))
        .apply(
            JdbcIO.<KV<Integer, String>>write()
                .withDataSourceConfiguration(DATA_SOURCE_CONFIGURATION)
                .withStatement(String.format("insert into %s values(?, ?)", tableName))
                .withRetryStrategy(
                    (JdbcIO.RetryStrategy)
                        e -> {
                          return "40XL1"
                              .equals(e.getSQLState()); // we fake a deadlock with a lock here
                        })
                .withPreparedStatementSetter(
                    (element, statement) -> {
                      statement.setInt(1, element.getKey());
                      statement.setString(2, element.getValue());
                    }));

    // starting a thread to perform the commit later, while the pipeline is running into the backoff
    final Thread commitThread =
        new Thread(
            () -> {
              while (true) {
                try {
                  Thread.sleep(500);
                  expectedLogs.verifyWarn("Deadlock detected, retrying");
                  break;
                } catch (AssertionError | java.lang.InterruptedException e) {
                  // nothing to do
                }
              }
              try {
                connection.commit();
              } catch (Exception e) {
                // nothing to do.
              }
            });

    commitThread.start();
    PipelineResult result = pipeline.run();
    commitThread.join();
    result.waitUntilFinish();

    // we verify that the backoff has been called thanks to the log message
    expectedLogs.verifyWarn("Deadlock detected, retrying");

    assertRowCount(DATA_SOURCE, tableName, 2);
  }