@Test(expected = ClassNotFoundException.class) public void testGetStreamingMessageParserClassThrowsClassNotFoundException() throws ClassNotFoundException { ConcurrentHashMap<String, String> concurrentHashMap = new ConcurrentHashMap<>(); concurrentHashMap.put("message.parser", "ev"); KafkaSource.getStreamingMessageParserClass(concurrentHashMap); }
public static Class<?> getStreamingMessageParserClass(Map<String, String> sourceProperties) throws ClassNotFoundException { String parserName = sourceProperties.get(PROP_MESSAGE_PARSER); String parserClazzName = DEF_MSSAGE_PARSER_CLAZZ; if (parserName != null) { parserClazzName = getParserClassName(parserName); } return Class.forName(parserClazzName); }
KafkaSource implements IStreamingSource { public static Class<?> getStreamingMessageParserClass(Map<String, String> sourceProperties) throws ClassNotFoundException { String parserName = sourceProperties.get(PROP_MESSAGE_PARSER); String parserClazzName = DEF_MSSAGE_PARSER_CLAZZ; if (parserName != null) { parserClazzName = getParserClassName(parserName); } return Class.forName(parserClazzName); } }
KafkaSource implements IStreamingSource { public static Class<?> getStreamingMessageParserClass(Map<String, String> sourceProperties) throws ClassNotFoundException { String parserName = sourceProperties.get(PROP_MESSAGE_PARSER); String parserClazzName = DEF_MSSAGE_PARSER_CLAZZ; if (parserName != null) { parserClazzName = getParserClassName(parserName); } return Class.forName(parserClazzName); }  }
KafkaSource implements IStreamingSource { public static Class<?> getStreamingMessageParserClass(Map<String, String> sourceProperties) throws ClassNotFoundException { String parserName = sourceProperties.get(PROP_MESSAGE_PARSER); String parserClazzName = DEF_MSSAGE_PARSER_CLAZZ; if (parserName != null) { parserClazzName = getParserClassName(parserName); } return Class.forName(parserClazzName); }  @Override StreamingTableSourceInfo load(String cubeName); @Override String getMessageTemplate(StreamingSourceConfig streamingSourceConfig); @Override IStreamingConnector createStreamingConnector(String cubeName, List<Partition> assignedPartitions,
            ConsumerStartProtocol startProtocol, StreamingSegmentManager streamingSegmentManager); @Override ISourcePositionHandler getSourcePositionHandler(); static Map<String, Object> getKafkaConf(Map<String, String> sourceProperties, KylinConfig kylinConfig); static Map<String, Object> getKafkaConf(Map<String, String> sourceProperties); static String getBootstrapServers(Map<String, String> sourceProperties); static String getTopicName(Map<String, String> sourceProperties); static Class<?> getStreamingMessageParserClass(Map<String, String> sourceProperties); static String getParserClassName(String parser); Map<Integer, Long> calConsumeLag(String cubeName, ISourcePosition currentPosition); }
KafkaSource implements IStreamingSource { public static Class<?> getStreamingMessageParserClass(Map<String, String> sourceProperties) throws ClassNotFoundException { String parserName = sourceProperties.get(PROP_MESSAGE_PARSER); String parserClazzName = DEF_MSSAGE_PARSER_CLAZZ; if (parserName != null) { parserClazzName = getParserClassName(parserName); } return Class.forName(parserClazzName); }  @Override StreamingTableSourceInfo load(String cubeName); @Override String getMessageTemplate(StreamingSourceConfig streamingSourceConfig); @Override IStreamingConnector createStreamingConnector(String cubeName, List<Partition> assignedPartitions,
            ConsumerStartProtocol startProtocol, StreamingSegmentManager streamingSegmentManager); @Override ISourcePositionHandler getSourcePositionHandler(); static Map<String, Object> getKafkaConf(Map<String, String> sourceProperties, KylinConfig kylinConfig); static Map<String, Object> getKafkaConf(Map<String, String> sourceProperties); static String getBootstrapServers(Map<String, String> sourceProperties); static String getTopicName(Map<String, String> sourceProperties); static Class<?> getStreamingMessageParserClass(Map<String, String> sourceProperties); static String getParserClassName(String parser); Map<Integer, Long> calConsumeLag(String cubeName, ISourcePosition currentPosition); static final String PROP_TOPIC; static final String PROP_BOOTSTRAP_SERVERS; static final String PROP_MESSAGE_PARSER; }