StreamKafkaP extends AbstractProcessor { @Override protected void init(@Nonnull Context context) { processorIndex = context.globalProcessorIndex(); totalParallelism = context.totalParallelism(); consumer = new KafkaConsumer<>(properties); assignPartitions(false); } StreamKafkaP(
            @Nonnull Properties properties,
            @Nonnull List<String> topics,
            @Nonnull FunctionEx<? super ConsumerRecord<K, V>, ? extends T> projectionFn,
            @Nonnull EventTimePolicy<? super T> eventTimePolicy
    ); @Override boolean isCooperative(); @Override boolean complete(); @Override void close(); @Override boolean saveToSnapshot(); @Override void restoreFromSnapshot(@Nonnull Object key, @Nonnull Object value); @Override boolean finishSnapshotRestore(); @Nonnull static SupplierEx<Processor> processorSupplier(
            @Nonnull Properties properties,
            @Nonnull List<String> topics,
            @Nonnull FunctionEx<? super ConsumerRecord<K, V>, ? extends T> projectionFn,
            @Nonnull EventTimePolicy<? super T> eventTimePolicy
    );  }
@Test public void when_eventsInAllPartitions_then_watermarkOutputImmediately() throws Exception { StreamKafkaP processor = createProcessor(properties(), 1, r -> entry(r.key(), r.value()), 10_000); TestOutbox outbox = new TestOutbox(new int[]{10}, 10); processor.init(outbox, new TestProcessorContext()); for (int i = 0; i < INITIAL_PARTITION_COUNT; i++) { Entry<Integer, String> event = entry(i + 100, Integer.toString(i)); System.out.println("produced event " + event); kafkaTestSupport.produce(topic1Name, i, null, event.getKey(), event.getValue()); if (i == INITIAL_PARTITION_COUNT - 1) { assertEquals(new Watermark(100 - LAG), consumeEventually(processor, outbox)); } assertEquals(event, consumeEventually(processor, outbox)); } }
@Test public void when_eventsInSinglePartition_then_watermarkAfterIdleTime() throws Exception { StreamKafkaP processor = createProcessor(properties(), 2, r -> entry(r.key(), r.value()), 10_000); TestOutbox outbox = new TestOutbox(new int[]{10}, 10); processor.init(outbox, new TestProcessorContext()); kafkaTestSupport.produce(topic1Name, 10, "foo"); assertEquals(entry(10, "foo"), consumeEventually(processor, outbox)); long time1 = System.nanoTime(); assertEquals(new Watermark(10 - LAG), consumeEventually(processor, outbox)); long time2 = System.nanoTime(); long elapsedMs = NANOSECONDS.toMillis(time2 - time1); assertBetween("elapsed time", elapsedMs, 3000, 30_000); }
@Test public void when_partitionAdded_then_consumedFromBeginning() throws Exception { Properties properties = properties(); properties.setProperty("metadata.max.age.ms", "100"); StreamKafkaP processor = createProcessor(properties, 2, r -> entry(r.key(), r.value()), 10_000); TestOutbox outbox = new TestOutbox(new int[]{10}, 10); processor.init(outbox, new TestProcessorContext()); kafkaTestSupport.produce(topic1Name, 0, "0"); assertEquals(entry(0, "0"), consumeEventually(processor, outbox)); kafkaTestSupport.setPartitionCount(topic1Name, INITIAL_PARTITION_COUNT + 2); Thread.sleep(1000); kafkaTestSupport.resetProducer(); boolean somethingInPartition1 = false; for (int i = 1; i < 11; i++) { Future<RecordMetadata> future = kafkaTestSupport.produce(topic1Name, i, Integer.toString(i)); RecordMetadata recordMetadata = future.get(); System.out.println("Entry " + i + " produced to partition " + recordMetadata.partition()); somethingInPartition1 |= recordMetadata.partition() == 1; } assertTrue("nothing was produced to partition-1", somethingInPartition1); Set<Object> receivedEvents = new HashSet<>(); for (int i = 1; i < 11; i++) { try { receivedEvents.add(consumeEventually(processor, outbox)); } catch (AssertionError e) { throw new AssertionError("Unable to receive 10 items, events so far: " + receivedEvents); } } assertEquals(range(1, 11).mapToObj(i -> entry(i, Integer.toString(i))).collect(toSet()), receivedEvents); }
@Test public void when_customProjection_then_used() throws Exception { StreamKafkaP processor = createProcessor(properties(), 2, r -> r.key() + "=" + r.value(), 10_000); TestOutbox outbox = new TestOutbox(new int[]{10}, 10); processor.init(outbox, new TestProcessorContext()); kafkaTestSupport.produce(topic1Name, 0, "0"); assertEquals("0=0", consumeEventually(processor, outbox)); }