You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "coniferous-cmd (via GitHub)" <gi...@apache.org> on 2026/06/17 06:36:44 UTC
[I] [求助] Pipe 开启后必须手动执行 flush 才能同步数据 [iotdb]
coniferous-cmd opened a new issue, #17972:
URL: https://github.com/apache/iotdb/issues/17972
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/iotdb/issues) and found nothing similar.
### Motivation
### IoTDB 版本
```text
IoTDB 2.0.8
```
### 问题
Pipe 已正常创建并启动，数据持续写入源端数据库。
但是新写入的数据不会立即通过 Pipe 同步到目标端，只有在手动执行：
```sql
flush
```
之后，之前写入的数据才会被同步。
表现上看，Pipe 似乎依赖 TsFile Flush 之后才能感知到新增数据，而不是实时同步内存中的写入数据。
### 复现步骤
1. 创建 Pipe
```sql
package com.xlkh.tsdb.iotdb.plugin.processor;
import com.xlkh.tsdb.iotdb.Globals;
import com.xlkh.tsdb.iotdb.util.TableRowUtil;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.iotdb.pipe.api.PipeProcessor;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConsoleProcessor implements PipeProcessor {
private static final Logger log = LoggerFactory.getLogger(ConsoleProcessor.class);
private static final ConcurrentSkipListMap<Long, Map<String, String>> payloads = new ConcurrentSkipListMap<>();
private static final AtomicInteger counter = new AtomicInteger(0);
@Override
public void validate(PipeParameterValidator pipeParameterValidator) throws Exception {
}
@Override
public void customize(PipeParameters pipeParameters, PipeProcessorRuntimeConfiguration pipeProcessorRuntimeConfiguration) throws Exception {
counter.set(Integer.parseInt(pipeParameters.getAttribute().getOrDefault("count", "200000")));
}
@Override
public void process(TabletInsertionEvent tabletInsertionEvent, EventCollector eventCollector)
throws Exception {
tabletInsertionEvent.processTablet(((tablet, collector) -> {
TableRowUtil.toMap(tablet).forEach((ts, fields) ->
payloads.merge(ts, fields, (oldVal, newVal) -> {
oldVal.putAll(newVal);
return oldVal;
}));
}));
Long timestamp = payloads.firstKey();
log.info("payloads size: {} - {}", timestamp, payloads.get(timestamp).size());
if (payloads.get(timestamp).size() >= counter.get()) {
Entry<Long, Map<String, String>> firstEntry = payloads.pollFirstEntry();
if (firstEntry != null) {
Globals.EXECUTOR.submit(() -> {
log.info("time: {} - {}", timestamp, System.currentTimeMillis());
log.info("payloads size: {}", payloads.size());
log.info("{}", Globals.gson.toJson(firstEntry.getValue()));
});
}
}
eventCollector.collect(tabletInsertionEvent);
}
@Override
public void process(Event event, EventCollector eventCollector) throws Exception {
eventCollector.collect(event);
}
@Override
public void close() throws Exception {
payloads.clear();
counter.set(200000);
}
}
```
2. 启动 Pipe
```sql
DROP PIPE IF EXISTS console_pipe;
DROP PIPEPLUGIN IF EXISTS console_processor;
CREATE
PIPEPLUGIN console_processor AS 'com.xlkh.tsdb.iotdb.plugin.processor.ConsoleProcessor'
USING URI 'file:///home/iotdb/apache-iotdb-2.0.8-all-bin/ext/pipe/iotdb-plugin-ext.jar';
CREATE
PIPE console_pipe
WITH SOURCE (
'source' = 'iotdb-source',
'source.mode' = 'stream',
'source.pattern' = 'root.s1_5w',
'history.enable'='false',
'realtime.enable'='true',
)
WITH PROCESSOR (
'count' = '200000',
'processor' = 'console_processor',
)
WITH SINK (
'sink' = 'do-nothing-sink',
);
```
3. 持续写入数据
```sql
INSERT INTO root.test.d1(time, s1) VALUES (NOW(), 190);
```
4. 执行
```sql
flush
```
发现之前未同步的数据立即开始同步。
Pipe 在 Streaming 模式下能够实时同步写入数据，无需手动执行 `flush`。
### 补充信息
* Pipe 状态正常
* Pipe 未报错
* 问题可以稳定复现
* 已确认源端数据写入成功
* 希望确认这是配置问题还是 Pipe 的已知缺陷
### Solution
_No response_
### Alternatives
_No response_
### Are you willing to submit a PR?
- [ ] I'm willing to submit a PR!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [求助] Pipe 开启后必须手动执行 flush 才能同步数据 [iotdb]
Posted by "JackieTien97 (via GitHub)" <gi...@apache.org>.
JackieTien97 closed issue #17972: [求助] Pipe 开启后必须手动执行 flush 才能同步数据
URL: https://github.com/apache/iotdb/issues/17972
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [求助] Pipe 开启后必须手动执行 flush 才能同步数据 [iotdb]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on issue #17972:
URL: https://github.com/apache/iotdb/issues/17972#issuecomment-4727191134
https://github.com/apache/iotdb/issues/17846
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org