KafkaConsumerInputMeta extends BaseStreamStepMeta implements StepMetaInterface { @Override public void loadXML( Node stepnode, List<DatabaseMeta> databases, IMetaStore metaStore ) { readData( stepnode ); } KafkaConsumerInputMeta(); @Override void loadXML( Node stepnode, List<DatabaseMeta> databases, IMetaStore metaStore ); @Override void setDefault(); @Override void readRep( Repository rep, IMetaStore metaStore, ObjectId objectId, List<DatabaseMeta> databases ); @Override void saveRep( Repository rep, IMetaStore metaStore, ObjectId transId, ObjectId stepId ); @Override RowMeta getRowMeta( String origin, VariableSpace space ); @Override StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int cnr, TransMeta tr,
                                Trans trans ); @Override StepDataInterface getStepData(); void setTopics( List<String> topics ); void addTopic( String topic ); void setConsumerGroup( String consumerGroup ); @Metaverse.Node ( name = KAFKA_SERVER_METAVERSE, type = KAFKA_SERVER_METAVERSE ) @Metaverse.Property ( name = KAFKA_SERVER_METAVERSE, parentNodeName = KAFKA_SERVER_METAVERSE ) String getBootstrapServers(); @Metaverse.Node ( name = KAFKA_TOPIC_METAVERSE, type = KAFKA_TOPIC_METAVERSE, link = LINK_READBY ) @Metaverse.Property ( name = TOPIC, parentNodeName = KAFKA_TOPIC_METAVERSE ) List<String> getTopics(); String getConsumerGroup(); KafkaConsumerField getKeyField(); KafkaConsumerField getMessageField(); KafkaConsumerField getTopicField(); KafkaConsumerField getOffsetField(); KafkaConsumerField getPartitionField(); KafkaConsumerField getTimestampField(); String getDirectBootstrapServers(); void setKeyField( KafkaConsumerField keyField ); void setMessageField( KafkaConsumerField messageField ); void setTopicField( KafkaConsumerField topicField ); void setOffsetField( KafkaConsumerField offsetField ); void setPartitionField( KafkaConsumerField partitionField ); void setTimestampField( KafkaConsumerField timestampField ); void setDirectBootstrapServers( final String directBootstrapServers ); ConnectionType getConnectionType(); void setConnectionType( final ConnectionType connectionType ); @Override String getXML(); List<KafkaConsumerField> getFieldDefinitions(); KafkaFactory getKafkaFactory(); NamedClusterService getNamedClusterService(); String getClusterName(); void setClusterName( String clusterName ); MetastoreLocator getMetastoreLocator(); void setNamedClusterService( NamedClusterService namedClusterService ); void setAutoCommit( boolean autoCommit ); boolean isAutoCommit(); Optional<JaasConfigService> getJaasConfigService(); NamedClusterServiceLocator getNamedClusterServiceLocator(); void setNamedClusterServiceLocator( NamedClusterServiceLocator namedClusterServiceLocator ); void setMetastoreLocator( MetastoreLocator metastoreLocator ); void setConfig( Map<String, String> config ); Map<String, String> getConfig(); @Metaverse.Node ( name = KEY_FIELD_NAME, type = RESOURCE, link = LINK_INPUTS, nameFromValue = FALSE, subTransLink = SUBTRANS_INPUT ) @Metaverse.Property ( name = KEY_FIELD_NAME, parentNodeName = KEY_FIELD_NAME ) @Metaverse.NodeLink ( nodeName = KEY_FIELD_NAME, parentNodeName = KAFKA_TOPIC_METAVERSE, linkDirection = "OUT" ) String getKeyOutputName(); @Metaverse.Node ( name = MSG_FIELD_NAME, type = RESOURCE, link = LINK_INPUTS, nameFromValue = FALSE, subTransLink = SUBTRANS_INPUT ) @Metaverse.NodeLink ( nodeName = MSG_FIELD_NAME, parentNodeName = KAFKA_TOPIC_METAVERSE, linkDirection = "OUT" ) String getMessageOutputName(); @Metaverse.Node ( name = TOPIC_FIELD_NAME, type = RESOURCE, link = LINK_INPUTS, nameFromValue = FALSE, subTransLink = SUBTRANS_INPUT ) @Metaverse.NodeLink ( nodeName = TOPIC_FIELD_NAME, parentNodeName = KAFKA_TOPIC_METAVERSE, linkDirection = "OUT" ) String getTopicOutputName(); @Metaverse.Node ( name = PARTITION_FIELD_NAME, type = RESOURCE, link = LINK_INPUTS, nameFromValue = FALSE, subTransLink = SUBTRANS_INPUT ) @Metaverse.NodeLink ( nodeName = PARTITION_FIELD_NAME, parentNodeName = KAFKA_TOPIC_METAVERSE, linkDirection = "OUT" ) String getPartitionOutputName(); @Metaverse.Node ( name = OFFSET_FIELD_NAME, type = RESOURCE, link = LINK_INPUTS, nameFromValue = FALSE, subTransLink = SUBTRANS_INPUT ) @Metaverse.NodeLink ( nodeName = OFFSET_FIELD_NAME, parentNodeName = KAFKA_TOPIC_METAVERSE, linkDirection = "OUT" ) String getOffsetOutputName(); @Metaverse.Node ( name = TIMESTAMP_FIELD_NAME, type = RESOURCE, link = LINK_INPUTS, nameFromValue = FALSE, subTransLink = SUBTRANS_INPUT ) @Metaverse.NodeLink ( nodeName = TIMESTAMP_FIELD_NAME, parentNodeName = KAFKA_TOPIC_METAVERSE, linkDirection = "OUT" ) String getTimestampOutputName(); static final String CLUSTER_NAME; static final String TOPIC; static final String CONSUMER_GROUP; static final String TRANSFORMATION_PATH; static final String BATCH_SIZE; static final String BATCH_DURATION; static final String PREFETCH_MESSAGE_COUNT; static final String CONNECTION_TYPE; static final String DIRECT_BOOTSTRAP_SERVERS; static final String ADVANCED_CONFIG; static final String CONFIG_OPTION; static final String OPTION_PROPERTY; static final String OPTION_VALUE; static final String TOPIC_FIELD_NAME; static final String OFFSET_FIELD_NAME; static final String PARTITION_FIELD_NAME; static final String TIMESTAMP_FIELD_NAME; static final String OUTPUT_FIELD_TAG_NAME; static final String KAFKA_NAME_ATTRIBUTE; static final String TYPE_ATTRIBUTE; static final String AUTO_COMMIT; }
@Test public void nullDurationSizeLoadedAsEmptyString() throws Exception { KafkaConsumerInputMeta meta = new KafkaConsumerInputMeta(); String inputXml = " <step>\n" + " <name>Kafka Consumer</name>\n" + " <type>KafkaConsumerInput</type>\n" + " <description />\n" + " <distribute>Y</distribute>\n" + " <custom_distribution />\n" + " <copies>1</copies>\n" + " <partitioning>\n" + " <method>none</method>\n" + " <schema_name />\n" + " </partitioning>\n" + " <clusterName>some_cluster</clusterName>\n" + " <directBootstrapServers>some_host:123,some_other_host:456</directBootstrapServers>\n" + " <connectionType>CLUSTER</connectionType>\n" + " <topic>one</topic>\n" + " <consumerGroup>two</consumerGroup>\n" + " <transformationPath>/home/pentaho/myKafkaTransformation.ktr</transformationPath>\n" + " <SUB_STEP>Filter</SUB_STEP>\n" + " <batchSize/>\n" + " <batchDuration/>\n" + " <OutputField kafkaName=\"key\" type=\"String\">three</OutputField>\n" + " <OutputField kafkaName=\"message\" type=\"String\">four</OutputField>\n" + " <OutputField kafkaName=\"topic\" type=\"String\">five</OutputField>\n" + " <OutputField kafkaName=\"partition\" type=\"Integer\">six</OutputField>\n" + " <OutputField kafkaName=\"offset\" type=\"Integer\">seven</OutputField>\n" + " <OutputField kafkaName=\"timestamp\" type=\"Integer\">eight</OutputField>\n" + " <advancedConfig>\n" + " <option property=\"advanced.property1\" value=\"advancedPropertyValue1\"></option>\n" + " <option property=\"advanced.property2\" value=\"advancedPropertyValue2\"></option>\n" + " </advancedConfig>\n" + " <cluster_schema />\n" + " <remotesteps>\n" + " <input>\n" + " </input>\n" + " <output>\n" + " </output>\n" + " </remotesteps>\n" + " <GUI>\n" + " <xloc>208</xloc>\n" + " <yloc>80</yloc>\n" + " <draw>Y</draw>\n" + " </GUI>\n" + " </step>\n"; Node node = XMLHandler.loadXMLString( inputXml ).getFirstChild(); meta.loadXML( node, Collections.emptyList(), metastore ); assertEquals( "", meta.getBatchDuration() ); assertEquals( "", meta.getBatchSize() ); }