content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.streams.test;
  
  import org.apache.kafka.common.serialization.Serdes;
  import org.apache.kafka.streams.StreamsConfig;
  import org.apache.kafka.streams.processor.PunctuationType;
  import org.apache.kafka.streams.processor.Punctuator;
  import org.apache.kafka.streams.processor.TaskId;
  import org.apache.kafka.streams.processor.api.MockProcessorContext;
  import org.apache.kafka.streams.processor.api.MockProcessorContext.CapturedForward;
  import org.apache.kafka.streams.processor.api.Processor;
  import org.apache.kafka.streams.processor.api.ProcessorContext;
  import org.apache.kafka.streams.processor.api.Record;
  import org.apache.kafka.streams.processor.api.RecordMetadata;
  import org.apache.kafka.streams.state.KeyValueStore;
  import org.apache.kafka.streams.state.StoreBuilder;
  import org.apache.kafka.streams.state.Stores;
  
  import org.junit.jupiter.api.Test;
  
  import java.io.File;
  import java.time.Duration;
  import java.util.List;
  import java.util.Optional;
  import java.util.Properties;
  
  import static java.util.Arrays.asList;
  import static java.util.Collections.singletonList;
  import static org.apache.kafka.common.utils.Utils.mkEntry;
  import static org.apache.kafka.common.utils.Utils.mkMap;
  import static org.apache.kafka.common.utils.Utils.mkProperties;
  import static org.hamcrest.MatcherAssert.assertThat;
  import static org.hamcrest.Matchers.empty;
  import static org.hamcrest.Matchers.is;
  
  public class MockProcessorContextAPITest {
      @Test
      public void shouldCaptureOutputRecords() {
          final Processor<String, Long, String, Long> processor = new Processor<>() {
              private ProcessorContext<String, Long> context;
  
              @Override
              public void init(final ProcessorContext<String, Long> context) {
                  this.context = context;
              }
  
              @Override
              public void process(final Record<String, Long> record) {
                  final String key = record.key();
                  final Long value = record.value();
                  context.forward(record.withKey(key + value).withValue(key.length() + value));
              }
          };
  
          final MockProcessorContext<String, Long> context = new MockProcessorContext<>();
          processor.init(context);
  
          processor.process(new Record<>("foo", 5L, 0L));
          processor.process(new Record<>("barbaz", 50L, 0L));
  
          final List<CapturedForward<? extends String, ? extends Long>> actual = context.forwarded();
          final List<CapturedForward<String, Long>> expected = asList(
              new CapturedForward<>(new Record<>("foo5", 8L, 0L)),
              new CapturedForward<>(new Record<>("barbaz50", 56L, 0L))
          );
          assertThat(actual, is(expected));
  
          context.resetForwards();
  
          assertThat(context.forwarded(), empty());
      }
  
      @Test
      public void shouldCaptureRecordsOutputToChildByName() {
          final Processor<String, Long, String, Long> processor = new Processor<>() {
              private ProcessorContext<String, Long> context;
  
              @Override
              public void process(final Record<String, Long> record) {
                  final String key = record.key();
                  final Long value = record.value();
                  if (count == 0) {
                      context.forward(new Record<>("start", -1L, 0L)); // broadcast
                  }
                  final String toChild = count % 2 == 0 ? "george" : "pete";
                  context.forward(new Record<>(key + value, key.length() + value, 0L), toChild);
                  count++;
              }
  
              @Override
              public void init(final ProcessorContext<String, Long> context) {
                  this.context = context;
              }
  
              private int count = 0;
  
          };
  
          final MockProcessorContext<String, Long> context = new MockProcessorContext<>();
  
          processor.init(context);
  
          processor.process(new Record<>("foo", 5L, 0L));
          processor.process(new Record<>("barbaz", 50L, 0L));
  
          {
              final List<CapturedForward<? extends String, ? extends Long>> forwarded = context.forwarded();
              final List<CapturedForward<? extends String, ? extends Long>> expected = asList(
                  new CapturedForward<>(new Record<>("start", -1L, 0L), Optional.empty()),
                  new CapturedForward<>(new Record<>("foo5", 8L, 0L), Optional.of("george")),
                  new CapturedForward<>(new Record<>("barbaz50", 56L, 0L), Optional.of("pete"))
              );
  
              assertThat(forwarded, is(expected));
          }
          {
              final List<CapturedForward<? extends String, ? extends Long>> forwarded = context.forwarded("george");
              final List<CapturedForward<? extends String, ? extends Long>> expected = asList(
                  new CapturedForward<>(new Record<>("start", -1L, 0L), Optional.empty()),
                  new CapturedForward<>(new Record<>("foo5", 8L, 0L), Optional.of("george"))
              );
  
              assertThat(forwarded, is(expected));
          }
          {
              final List<CapturedForward<? extends String, ? extends Long>> forwarded = context.forwarded("pete");
              final List<CapturedForward<? extends String, ? extends Long>> expected = asList(
                  new CapturedForward<>(new Record<>("start", -1L, 0L), Optional.empty()),
                  new CapturedForward<>(new Record<>("barbaz50", 56L, 0L), Optional.of("pete"))
              );
  
              assertThat(forwarded, is(expected));
          }
          {
              final List<CapturedForward<? extends String, ? extends Long>> forwarded = context.forwarded("steve");
              final List<CapturedForward<? extends String, ? extends Long>> expected = singletonList(
                  new CapturedForward<>(new Record<>("start", -1L, 0L))
              );
  
              assertThat(forwarded, is(expected));
          }
      }
  
      @Test
      public void shouldCaptureCommitsAndAllowReset() {
          final Processor<String, Long, Void, Void> processor = new Processor<>() {
              private ProcessorContext<Void, Void> context;
              private int count = 0;
  
              @Override
              public void init(final ProcessorContext<Void, Void> context) {
                  this.context = context;
              }
  
              @Override
              public void process(final Record<String, Long> record) {
                  if (++count > 2) {
                      context.commit();
                  }
              }
          };
  
          final MockProcessorContext<Void, Void> context = new MockProcessorContext<>();
  
          processor.init(context);
  
          processor.process(new Record<>("foo", 5L, 0L));
          processor.process(new Record<>("barbaz", 50L, 0L));
  
          assertThat(context.committed(), is(false));
  
          processor.process(new Record<>("foobar", 500L, 0L));
  
          assertThat(context.committed(), is(true));
  
          context.resetCommit();
  
          assertThat(context.committed(), is(false));
      }
  
      @Test
      public void shouldStoreAndReturnStateStores() {
          final Processor<String, Long, Void, Void> processor = new Processor<>() {
              private ProcessorContext<Void, Void> context;
  
              @Override
              public void init(final ProcessorContext<Void, Void> context) {
                  this.context = context;
              }
  
              @Override
              public void process(final Record<String, Long> record) {
                  final String key = record.key();
                  final Long value = record.value();
                  final KeyValueStore<String, Long> stateStore = context.getStateStore("my-state");
  
                  stateStore.put(key, (stateStore.get(key) == null ? 0 : stateStore.get(key)) + value);
                  stateStore.put("all", (stateStore.get("all") == null ? 0 : stateStore.get("all")) + value);
              }
  
          };
  
          final MockProcessorContext<Void, Void> context = new MockProcessorContext<>();
  
          final StoreBuilder<KeyValueStore<String, Long>> storeBuilder = Stores.keyValueStoreBuilder(
              Stores.inMemoryKeyValueStore("my-state"),
              Serdes.String(),
              Serdes.Long()).withLoggingDisabled();
  
          final KeyValueStore<String, Long> store = storeBuilder.build();
  
          store.init(context.getStateStoreContext(), store);
  
          processor.init(context);
  
          processor.process(new Record<>("foo", 5L, 0L));
          processor.process(new Record<>("bar", 50L, 0L));
  
          assertThat(store.get("foo"), is(5L));
          assertThat(store.get("bar"), is(50L));
          assertThat(store.get("all"), is(55L));
      }
  
  
      @Test
      public void shouldCaptureApplicationAndRecordMetadata() {
          final Properties config = mkProperties(
              mkMap(
                  mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, "testMetadata"),
                  mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "")
              )
          );
  
          final Processor<String, Object, String, Object> processor = new Processor<>() {
              private ProcessorContext<String, Object> context;
  
              @Override
              public void init(final ProcessorContext<String, Object> context) {
                  this.context = context;
              }
  
              @Override
              public void process(final Record<String, Object> record) {
                  context.forward(new Record<String, Object>("appId", context.applicationId(), 0L));
                  context.forward(new Record<String, Object>("taskId", context.taskId(), 0L));
  
                  if (context.recordMetadata().isPresent()) {
                      final RecordMetadata recordMetadata = context.recordMetadata().get();
                      context.forward(new Record<String, Object>("topic", recordMetadata.topic(), 0L));
                      context.forward(new Record<String, Object>("partition", recordMetadata.partition(), 0L));
                      context.forward(new Record<String, Object>("offset", recordMetadata.offset(), 0L));
                  }
  
                  context.forward(new Record<String, Object>("record", record, 0L));
              }
          };
  
          final MockProcessorContext<String, Object> context = new MockProcessorContext<>(config);
          processor.init(context);
  
          processor.process(new Record<>("foo", 5L, 0L));
          {
              final List<CapturedForward<? extends String, ?>> forwarded = context.forwarded();
              final List<CapturedForward<? extends String, ?>> expected = asList(
                  new CapturedForward<>(new Record<>("appId", "testMetadata", 0L)),
                  new CapturedForward<>(new Record<>("taskId", new TaskId(0, 0), 0L)),
                  new CapturedForward<>(new Record<>("record", new Record<>("foo", 5L, 0L), 0L))
              );
              assertThat(forwarded, is(expected));
          }
          context.resetForwards();
          context.setRecordMetadata("t1", 0, 0L);
          processor.process(new Record<>("foo", 5L, 0L));
          {
              final List<CapturedForward<? extends String, ?>> forwarded = context.forwarded();
              final List<CapturedForward<? extends String, ?>> expected = asList(
                  new CapturedForward<>(new Record<>("appId", "testMetadata", 0L)),
                  new CapturedForward<>(new Record<>("taskId", new TaskId(0, 0), 0L)),
                  new CapturedForward<>(new Record<>("topic", "t1", 0L)),
                  new CapturedForward<>(new Record<>("partition", 0, 0L)),
                  new CapturedForward<>(new Record<>("offset", 0L, 0L)),
                  new CapturedForward<>(new Record<>("record", new Record<>("foo", 5L, 0L), 0L))
              );
              assertThat(forwarded, is(expected));
          }
      }
  
      @Test
      public void shouldCapturePunctuator() {
          final Processor<String, Long, Void, Void> processor = new Processor<>() {
              @Override
              public void init(final ProcessorContext<Void, Void> context) {
                  context.schedule(
                      Duration.ofSeconds(1L),
                      PunctuationType.WALL_CLOCK_TIME,
                      timestamp -> context.commit()
                  );
              }
  
              @Override
              public void process(final Record<String, Long> record) {}
          };
  
          final MockProcessorContext<Void, Void> context = new MockProcessorContext<>();
  
          processor.init(context);
  
          final MockProcessorContext.CapturedPunctuator capturedPunctuator = context.scheduledPunctuators().get(0);
          assertThat(capturedPunctuator.getInterval(), is(Duration.ofMillis(1000L)));
          assertThat(capturedPunctuator.getType(), is(PunctuationType.WALL_CLOCK_TIME));
          assertThat(capturedPunctuator.cancelled(), is(false));
  
          final Punctuator punctuator = capturedPunctuator.getPunctuator();
          assertThat(context.committed(), is(false));
          punctuator.punctuate(1234L);
          assertThat(context.committed(), is(true));
      }
  
      @SuppressWarnings("resource")
      @Test
      public void fullConstructorShouldSetAllExpectedAttributes() {
          final Properties config = new Properties();
          config.put(StreamsConfig.APPLICATION_ID_CONFIG, "testFullConstructor");
          config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "");
          config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());
          config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.LongSerde.class.getName());
  
          final File dummyFile = new File("");
          final MockProcessorContext<Void, Void> context =
              new MockProcessorContext<>(config, new TaskId(1, 1), dummyFile);
  
          assertThat(context.applicationId(), is("testFullConstructor"));
          assertThat(context.taskId(), is(new TaskId(1, 1)));
          assertThat(context.appConfigs().get(StreamsConfig.APPLICATION_ID_CONFIG), is("testFullConstructor"));
          assertThat(context.appConfigsWithPrefix("application.").get("id"), is("testFullConstructor"));
          assertThat(context.keySerde().getClass(), is(Serdes.StringSerde.class));
          assertThat(context.valueSerde().getClass(), is(Serdes.LongSerde.class));
          assertThat(context.stateDir(), is(dummyFile));
      }
  }
  