@Test @Ignore public void testKafkaSourceSingleConsumerReadsAllMessagesInOrderFromSinglePartition() throws InterruptedException { String testTopic = "testTopic" + topicNum.incrementAndGet(); int numPartitions = 1; kafkaServer.createTopic(testTopic, numPartitions); int numMessages = 1; AtomicInteger counter = new AtomicInteger(); Schedulers.newThread().createWorker().schedulePeriodically(() -> { ProducerRecord<String, String> keyedMessage = new ProducerRecord<>(testTopic, "{\"messageNum\":" + counter.incrementAndGet() + "}"); kafkaServer.sendMessages(keyedMessage); }, 1, 1, TimeUnit.SECONDS); LocalJobExecutorNetworked.execute(new QueryableKafkaSourceJob().getJobInstance(), new Parameter(KafkaSourceParameters.TOPIC, testTopic), new Parameter(KafkaSourceParameters.NUM_KAFKA_CONSUMER_PER_WORKER, "1"), new Parameter(KafkaSourceParameters.PARSER_TYPE, ParserType.SIMPLE_JSON.getPropName()), new Parameter(KafkaSourceParameters.PARSE_MSG_IN_SOURCE, "true"), new Parameter(KafkaSourceParameters.PREFIX + ConsumerConfig.GROUP_ID_CONFIG, "QueryableKafkaSourceLocal"), new Parameter(KafkaSourceParameters.PREFIX + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"), new Parameter(KafkaSourceParameters.PREFIX + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")); final CountDownLatch latch = new CountDownLatch(numMessages); assertTrue("timed out waiting to get all messages from Kafka", latch.await(10, TimeUnit.MINUTES)); kafkaServer.deleteTopic(testTopic); }
@Override public Job<TaggedData> getJobInstance() { KafkaSource kafkaSource = new KafkaSource(new DefaultRegistry()); return MantisJob .source(kafkaSource) .stage(getAckableTaggingStage(), CustomizedAutoAckTaggingStage.config()) .sink(new TaggedDataSourceSink(new QueryRequestPreProcessor(), new QueryRequestPostProcessor())) .create(); }
QueryableKafkaSourceJob extends MantisJobProvider<TaggedData> { @Override public Job<TaggedData> getJobInstance() { KafkaSource kafkaSource = new KafkaSource(new DefaultRegistry()); return MantisJob .source(kafkaSource) .stage(getAckableTaggingStage(), CustomizedAutoAckTaggingStage.config()) .sink(new TaggedDataSourceSink(new QueryRequestPreProcessor(), new QueryRequestPostProcessor())) .create(); } }
QueryableKafkaSourceJob extends MantisJobProvider<TaggedData> { @Override public Job<TaggedData> getJobInstance() { KafkaSource kafkaSource = new KafkaSource(new DefaultRegistry()); return MantisJob .source(kafkaSource) .stage(getAckableTaggingStage(), CustomizedAutoAckTaggingStage.config()) .sink(new TaggedDataSourceSink(new QueryRequestPreProcessor(), new QueryRequestPostProcessor())) .create(); }  }
QueryableKafkaSourceJob extends MantisJobProvider<TaggedData> { @Override public Job<TaggedData> getJobInstance() { KafkaSource kafkaSource = new KafkaSource(new DefaultRegistry()); return MantisJob .source(kafkaSource) .stage(getAckableTaggingStage(), CustomizedAutoAckTaggingStage.config()) .sink(new TaggedDataSourceSink(new QueryRequestPreProcessor(), new QueryRequestPostProcessor())) .create(); }  @Override Job<TaggedData> getJobInstance(); static void main(String[] args); }
QueryableKafkaSourceJob extends MantisJobProvider<TaggedData> { @Override public Job<TaggedData> getJobInstance() { KafkaSource kafkaSource = new KafkaSource(new DefaultRegistry()); return MantisJob .source(kafkaSource) .stage(getAckableTaggingStage(), CustomizedAutoAckTaggingStage.config()) .sink(new TaggedDataSourceSink(new QueryRequestPreProcessor(), new QueryRequestPostProcessor())) .create(); }  @Override Job<TaggedData> getJobInstance(); static void main(String[] args);  }