luckydarnell commented on issue #108:
URL: https://github.com/apache/rocketmq-flink/issues/108#issuecomment-1882337267
I have tried several times and found that **_RocketMQSourceFunction_**
interface can run normally。so I guess there are some bugs in the
**_RocketMQSource_** interface。
+ code
```
public static RocketMQSourceFunction<TransData>
createMqsource2(SourceCfg cfg, DataSourceCfg dataSource) {
// 构造mq地址
String[] hostNameSplit =
dataSource.getHostname().split(SymbolConstant.SEPARATOR_SEMI_COLON);
String addr = null;
for (String s : hostNameSplit) {
String joinStr = s + SymbolConstant.SEPARATOR_COLON +
dataSource.getPort();
addr = StringUtils.isEmpty(addr) ? joinStr : addr +
SymbolConstant.SEPARATOR_SEMI_COLON + joinStr;
}
Properties consumerProps = new Properties();
consumerProps.setProperty(RocketMQConfig.NAME_SERVER_ADDR, addr);
consumerProps.setProperty(RocketMQConfig.CONSUMER_GROUP,
cfg.getTag());
consumerProps.setProperty(RocketMQConfig.CONSUMER_TOPIC,
cfg.getTopic());
KeyValueDeserializationSchema<TransData> deserializationSchema =
MqDeserializationSchemaFactory.crateMqDeserializationSchema(cfg);
// 判断启动模式
OffsetResetStrategy startUpMode = cfg.getStartupMode() == null ||
cfg.getStartupMode().equals(StartupModeEnum.LATEST) ?
OffsetResetStrategy.LATEST : OffsetResetStrategy.EARLIEST;
RocketMQSourceFunction<TransData> source = new
RocketMQSourceFunction(deserializationSchema, consumerProps);
// use group offsets.
// If there is no committed offset,consumer would start from the
latest offset.
source.setStartFromGroupOffsets(startUpMode);
log.info("addr:{}, group:{}, tag:{}", addr, cfg.getConsumerGroup(),
cfg.getTag());
return source;
}
```
--
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: commits-unsubscr...@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
