otter支持kafka作为数据源

三种模式

前端修改

给canal增加参数，也就是修改canalparamter.

List<List<DataSourcing>> groupDbAddresses =  canal.getCanalParameter().getGroupDbAddresses();
if(canal.getCanalParameter().getSourcingType().isKafka()){
  String  topicList = canal.getCanalParameter().getKafkaTopicList();
  DataSourcing dataSourcing = groupDbAddresses.get(0).get(0);
  String host = dataSourcing.getDbAddress().getHostString();
  int port = dataSourcing.getDbAddress().getPort();
}

切记看下是不是没修改form.xml，不然提交没法验证来着。获取不到表单的数据

			<field name="localBinlogDirectory"
				displayName="本地localBinlog目录">
				<fm-validators:if test="sourcingType.value == 'LOCALBINLOG'">
					<fm-validators:required-validator>
						<message>数据源为LOCALBINLOG时必须填写${displayName}</message>
					</fm-validators:required-validator>
				</fm-validators:if>
			</field>
			<field name="kafkaTopicList"
				displayName="kafka的topic列表">
				<fm-validators:if test="sourcingType.value == 'KAFKA'">
					<fm-validators:required-validator>
						<message>数据源为KAFKA时必须填写${displayName}</message>
					</fm-validators:required-validator>
				</fm-validators:if>
			</field>
			<field name="positions" displayName="位点信息">
				<fm-validators:regexp-validator pattern="^(\{.*\}\;)+$">
					<message>${displayName} 不符合格式（{"journalName":"","position":0,"timestamp":0};）</message>
				</fm-validators:regexp-validator>
			</field>