[Fix][Connector-V2] Fix RocketMQ parallism read on work well #9777
Conversation
|
Hi, The CI is failing due to a formatting issue (Spotless check). |
|
cc @Carl-Zhou-CN as well. |
|
@xxyykkxx hi, it seems that the changes don't just work on the flink engine |
yes,just because this issue can be intuitively identified through Flink monitoring |
Hisoka-X
changed the title
I changed the title. Please continue review this PR @Carl-Zhou-CN . Thanks. |
Sure, this is what I should do |
| Supplier<RocketMQPartitionSplitReader> splitReader = | ||
| () -> new RocketMQPartitionSplitReader(this.metadata, readerContext); | ||
|
|
||
| RocketMQSourceFetcherManager kafkaSourceFetcherManager = |
There was a problem hiding this comment.
| RocketMQSourceFetcherManager kafkaSourceFetcherManager = | |
| RocketMQSourceFetcherManager rocketMQSourceFetcherManager = |
|
|
||
| return new RocketMqSourceReader( | ||
| elementsQueue, | ||
| kafkaSourceFetcherManager, |
There was a problem hiding this comment.
| kafkaSourceFetcherManager, | |
| rocketMQSourceFetcherManager, |
| outputCollector.output = collector; | ||
| reportMetrics(consumerRecord); | ||
| deserializationSchema.deserialize(consumerRecord.getBody(), outputCollector); | ||
| // consumerRecord.offset + 1 is the offset commit to Kafka and also the start offset |
There was a problem hiding this comment.
| // consumerRecord.offset + 1 is the offset commit to Kafka and also the start offset | |
| // consumerRecord.offset + 1 is the offset commit to RocketMQ and also the start offset |
| return recordsBySplits; | ||
| } | ||
| try { | ||
| messageExts = consumer.poll(pollTimeOut); |
There was a problem hiding this comment.
Can we add a parameter to control pollTimeOut?
|
|
||
| /** | ||
| * @author 02211659 bianxiang | ||
| * @date 2025-08-19 10:06:05 |
| public class RocketMqRecordEmitter | ||
| implements RecordEmitter<MessageExt, SeaTunnelRow, RocketMQPartitionSplitState> { | ||
|
|
||
| private static final Logger logger = LoggerFactory.getLogger(RocketMqRecordEmitter.class); |
| // Set `rocketmq.client.logUseSlf4j` to `true` to avoid create many | ||
| // `AsyncAppender-Dispatcher-Thread` | ||
| System.setProperty("rocketmq.client.logUseSlf4j", "true"); | ||
| initialRocketMQConsumer(config); |
There was a problem hiding this comment.
Wouldn't it be better if we reused the implementation within RocketMqAdminUtil here?
| Thread thread = new Thread(runnable); | ||
| thread.setDaemon(true); | ||
| thread.setName("RocketMq-messageQueue-dynamic-discovery"); | ||
| thread.setName("kafka-partition-dynamic-discovery"); |
| // initialization scenario | ||
| if (offset <= -2) { | ||
| throw new RuntimeException( | ||
| "An error occurred while fetching offset,please check up server's log"); |
There was a problem hiding this comment.
Will continue to maintain RocketMqConnectorException better?
| break; | ||
| default: | ||
| throw new IllegalArgumentException( | ||
| "current startMode is not supported" + metadata.getStartMode()); |
|
This pull request has been automatically marked as stale because it has not had recent activity for 120 days. It will be closed in 7 days if no further activity occurs. |
4 participants
Purpose of this pull request
Does this PR introduce any user-facing change?
How was this patch tested?
Check list
New License Guide