Reject schema output for successful records

public class OutputSink implements Serializable {
  ...
  @ElementListener
  public void onNext(@Input final JsonObject record,
          @Output("reject") final OutputEmitter<Reject> reject) throws SQLException {
      try {
          ...
          statement.execute();
      } catch (Exception e) {
          reject.emit(new Reject(e.getMessage(), record));
      }
  }

[statistics] connected
|
|
|
...
[statistics] disconnected