public <KIn, VIn, KOut, VOut> ProcessorSupplier<KIn, VIn, KOut, VOut> process(String spanName,
    ProcessorSupplier<KIn, VIn, KOut, VOut> processorSupplier) {
    return new TracingProcessorSupplier<>(this, spanName, processorSupplier);
  }
@Test void newProcessorSupplier_should_tag_app_id_and_task_id() {
    Processor<String, String, String, String> processor =
      fakeV2ProcessorSupplier.get();
    processor.init(processorV2ContextSupplier.get());
    processor.process(new Record<>(TEST_KEY, TEST_VALUE, new Date().getTime()));

    assertThat(spans.get(0).tags()).containsOnly(
      entry("kafka.streams.application.id", TEST_APPLICATION_ID),
      entry("kafka.streams.task.id", TEST_TASK_ID));
  }