感觉这样子也并不能做到增量同步，因为我并不能确定上次同步的数据成功同步还是说同步失败

确实，如果使用 http_poller 插件进行增量同步，无法确定上次同步的数据是否成功同步到 Elasticsearch。如果数据同步失败，则可能会出现重复数据或者数据丢失的情况。

为了解决这个问题，可以考虑使用 Logstash 的 jdbc_input 插件或 file_input 插件进行增量同步。这些插件可以跟踪上次同步的位置，并且可以在同步失败时进行重试或记录错误日志。

例如，使用 jdbc_input 插件进行增量同步，可以指定 last_run_metadata_path 配置项来存储上次同步的位置。在每次同步时，jdbc_input 插件会使用 SQL 查询获取最新的数据，并且使用 last_run_metadata_path 中存储的位置信息来指定开始查询的位置。

input {
  jdbc {
    jdbc_connection_string => "jdbc:mysql://localhost:3306/mydb"
    jdbc_user => "myuser"
    jdbc_password => "mypassword"
    jdbc_driver_library => "/path/to/mysql-connector-java.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    statement => "SELECT * FROM mytable WHERE timestamp > :sql_last_value"
    use_column_value => true
    tracking_column => "timestamp"
    last_run_metadata_path => "/path/to/last_run_metadata/file"
  }
}

在这个示例中，jdbc_input 插件会查询 mytable 表中 timestamp 列大于上次同步的时间戳的数据，并使用 last_run_metadata_path 中存储的位置信息来指定开始查询的位置。同时，使用 tracking_column 配置项指定用于跟踪位置的列名。

使用 file_input 插件进行增量同步也可以类似地指定 last_run_metadata_path 配置项来存储上次同步的位置。在每次同步时，file_input 插件会从指定的文件中读取上次同步的位置，并从该位置继续读取数据。

input {
  file {
    path => "/path/to/data/file"
    start_position => "end"
    sincedb_path => "/path/to/sincedb/file"
    last_run_metadata_path => "/path/to/last_run_metadata/file"
  }
}

在这个示例中，file_input 插件会从 /path/to/data/file 文件的末尾开始读取数据，并使用 last_run_metadata_path 中存储的位置信息来指定开始读取的位置。同时，使用 sincedb_path 配置项指定用于跟踪读取位置的文件路径。

使用 jdbc_input 插件或 file_input 插件进行增量同步可以解决 http_poller 插件无法确定同步状态的问题，同时还提供了更多的功能和控制选项。