@Test public void testGetStreamingMessageParserClass() throws ClassNotFoundException { ConcurrentHashMap<String, String> concurrentHashMap = new ConcurrentHashMap<>(); Class<?> clasz = KafkaSource.getStreamingMessageParserClass(concurrentHashMap); assertTrue(concurrentHashMap.isEmpty()); assertFalse(clasz.isEnum()); assertFalse(clasz.isSynthetic()); assertEquals("class org.apache.kylin.stream.source.kafka.TimedJsonStreamParser", clasz.toString()); assertFalse(clasz.isPrimitive()); assertFalse(clasz.isAnnotation()); assertFalse(clasz.isInterface()); assertFalse(clasz.isArray()); }
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; }