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.kstream.internals.foreignkeyjoin;
  
  import org.apache.kafka.streams.kstream.internals.Change;
  import org.apache.kafka.streams.kstream.internals.KTableValueGetter;
  import org.apache.kafka.streams.kstream.internals.KTableValueGetterSupplier;
  import org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionWrapper.Instruction;
  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.state.ValueAndTimestamp;
  
  import org.junit.jupiter.api.Test;
  
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  
  public class SubscriptionJoinProcessorSupplierTest {
      final Map<String, ValueAndTimestamp<String>> fks = Collections.singletonMap(
          "fk1", ValueAndTimestamp.make("foo", 1L)
      );
      final KTableValueGetterSupplier<String, String> valueGetterSupplier = valueGetterSupplier(fks);
      final Processor<CombinedKey<String, String>,
                      Change<ValueAndTimestamp<SubscriptionWrapper<String>>>,
                      String,
                      SubscriptionResponseWrapper<String>>
          processor = processor(valueGetterSupplier);
  
      @Test
      public void shouldDetectVersionChange() {
          // This test serves as a reminder to add new tests once we bump SubscriptionWrapper version.
          assertEquals(SubscriptionWrapper.VERSION_1, SubscriptionWrapper.CURRENT_VERSION);
      }
  
      @Test
      public void shouldDeleteKeyAndPropagateFKV0() {
          final MockProcessorContext<String, SubscriptionResponseWrapper<String>> context = new MockProcessorContext<>();
          processor.init(context);
  
          final SubscriptionWrapper<String> newValue = new SubscriptionWrapper<>(
              new long[]{1L},
              Instruction.DELETE_KEY_AND_PROPAGATE,
              "pk1",
              SubscriptionWrapper.VERSION_0,
              null
          );
          final Record<CombinedKey<String, String>, Change<ValueAndTimestamp<SubscriptionWrapper<String>>>> record =
              new Record<>(
                  new CombinedKey<>("fk1", "pk1"),
                  new Change<>(ValueAndTimestamp.make(newValue, 1L), null),
                  1L
              );
          processor.process(record);
          final List<CapturedForward<? extends String, ? extends SubscriptionResponseWrapper<String>>> forwarded = context.forwarded();
          assertEquals(1, forwarded.size());
          assertEquals(
              new Record<>(
                  "pk1",
                  new SubscriptionResponseWrapper<>(
                      newValue.hash(),
                      null,
                      null),
                  1L
              ),
              forwarded.get(0).record()
          );
      }
  
      @Test
      public void shouldDeleteKeyAndPropagateFKV1() {
          final MockProcessorContext<String, SubscriptionResponseWrapper<String>> context = new MockProcessorContext<>();
          processor.init(context);
  
          final SubscriptionWrapper<String> newValue = new SubscriptionWrapper<>(
              new long[]{1L},
              Instruction.DELETE_KEY_AND_PROPAGATE,
              "pk1",
              SubscriptionWrapper.VERSION_1,
              12
          );
          final Record<CombinedKey<String, String>, Change<ValueAndTimestamp<SubscriptionWrapper<String>>>> record =
              new Record<>(
                  new CombinedKey<>("fk1", "pk1"),
                  new Change<>(ValueAndTimestamp.make(newValue, 1L), null),
                  1L
              );
          processor.process(record);
          final List<CapturedForward<? extends String, ? extends SubscriptionResponseWrapper<String>>> forwarded = context.forwarded();
          assertEquals(1, forwarded.size());
          assertEquals(
              new Record<>(
                  "pk1",
                  new SubscriptionResponseWrapper<>(
                      newValue.hash(),
                      null,
                      12
                  ),
                  1L
              ),
              forwarded.get(0).record()
          );
      }
  
      @Test
      public void shouldPropagateOnlyIfFKAvailableV0() {
          final MockProcessorContext<String, SubscriptionResponseWrapper<String>> context = new MockProcessorContext<>();
          processor.init(context);
  
          final SubscriptionWrapper<String> newValue = new SubscriptionWrapper<>(
              new long[]{1L},
              Instruction.PROPAGATE_ONLY_IF_FK_VAL_AVAILABLE,
              "pk1",
              SubscriptionWrapper.VERSION_0,
              null
          );
          final Record<CombinedKey<String, String>, Change<ValueAndTimestamp<SubscriptionWrapper<String>>>> record =
              new Record<>(
                  new CombinedKey<>("fk1", "pk1"),
                  new Change<>(ValueAndTimestamp.make(newValue, 1L), null),
                  1L
          );
          processor.process(record);
          final List<CapturedForward<? extends String, ? extends SubscriptionResponseWrapper<String>>> forwarded = context.forwarded();
          assertEquals(1, forwarded.size());
          assertEquals(
              new Record<>(
                  "pk1",
                  new SubscriptionResponseWrapper<>(
                      newValue.hash(),
                      "foo",
                      null
                  ),
                  1L
              ),
              forwarded.get(0).record()
          );
      }
  
      @Test
      public void shouldPropagateOnlyIfFKAvailableV1() {
          final MockProcessorContext<String, SubscriptionResponseWrapper<String>> context = new MockProcessorContext<>();
          processor.init(context);
  
          final SubscriptionWrapper<String> newValue = new SubscriptionWrapper<>(
              new long[]{1L},
              Instruction.PROPAGATE_ONLY_IF_FK_VAL_AVAILABLE,
              "pk1",
              SubscriptionWrapper.VERSION_1,
              12
          );
          final Record<CombinedKey<String, String>, Change<ValueAndTimestamp<SubscriptionWrapper<String>>>> record =
              new Record<>(
                  new CombinedKey<>("fk1", "pk1"),
                  new Change<>(ValueAndTimestamp.make(newValue, 1L), null),
                  1L
              );
          processor.process(record);
          final List<CapturedForward<? extends String, ? extends SubscriptionResponseWrapper<String>>> forwarded = context.forwarded();
          assertEquals(1, forwarded.size());
          assertEquals(
              new Record<>(
                  "pk1",
                  new SubscriptionResponseWrapper<>(
                      newValue.hash(),
                      "foo",
                       12
                  ),
                  1L
              ),
              forwarded.get(0).record());
      }
  
      @Test
      public void shouldPropagateNullIfNoFKAvailableV0() {
          final MockProcessorContext<String, SubscriptionResponseWrapper<String>> context = new MockProcessorContext<>();
          processor.init(context);
  
          final SubscriptionWrapper<String> newValue = new SubscriptionWrapper<>(
              new long[]{1L},
              Instruction.PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE,
              "pk1",
              SubscriptionWrapper.VERSION_0,
              null
          );
          Record<CombinedKey<String, String>, Change<ValueAndTimestamp<SubscriptionWrapper<String>>>> record =
              new Record<>(
                  new CombinedKey<>("fk1", "pk1"),
                  new Change<>(ValueAndTimestamp.make(newValue, 1L), null),
                  1L
              );
          processor.process(record);
          // propagate matched FK
          List<CapturedForward<? extends String, ? extends SubscriptionResponseWrapper<String>>> forwarded = context.forwarded();
          assertEquals(1, forwarded.size());
          assertEquals(
              new Record<>(
                  "pk1",
                  new SubscriptionResponseWrapper<>(
                      newValue.hash(),
                      "foo",
                      null
                  ),
                  1L
              ),
              forwarded.get(0).record());
  
          record = new Record<>(
                  new CombinedKey<>("fk9000", "pk1"),
                  new Change<>(ValueAndTimestamp.make(newValue, 1L), null),
                  1L
              );
          processor.process(record);
          // propagate null if there is no match
          forwarded = context.forwarded();
          assertEquals(2, forwarded.size());
          assertEquals(
              new Record<>(
                  "pk1",
                  new SubscriptionResponseWrapper<>(
                      newValue.hash(),
                      null,
                      null
                  ),
                  1L
              ),
              forwarded.get(1).record());
      }
  
      @Test
      public void shouldPropagateNullIfNoFKAvailableV1() {
          final MockProcessorContext<String, SubscriptionResponseWrapper<String>> context = new MockProcessorContext<>();
          processor.init(context);
  
          final SubscriptionWrapper<String> newValue = new SubscriptionWrapper<>(
              new long[]{1L},
              Instruction.PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE,
              "pk1",
              SubscriptionWrapper.VERSION_1,
              12);
          Record<CombinedKey<String, String>, Change<ValueAndTimestamp<SubscriptionWrapper<String>>>> record =
              new Record<>(
                  new CombinedKey<>("fk1", "pk1"),
                  new Change<>(ValueAndTimestamp.make(newValue, 1L), null),
                  1L
              );
          processor.process(record);
          List<CapturedForward<? extends String, ? extends SubscriptionResponseWrapper<String>>> forwarded = context.forwarded();
          assertEquals(1, forwarded.size());
          assertEquals(
              new Record<>(
                  "pk1",
                  new SubscriptionResponseWrapper<>(
                      newValue.hash(),
                      "foo",
                      12
                  ),
                  1L
              ),
              forwarded.get(0).record());
  
          record = new Record<>(
              new CombinedKey<>("fk9000", "pk1"),
              new Change<>(ValueAndTimestamp.make(newValue, 1L), null),
              1L
          );
          processor.process(record);
          // propagate null if there is no match
          forwarded = context.forwarded();
          assertEquals(2, forwarded.size());
          assertEquals(
              new Record<>(
                  "pk1",
                  new SubscriptionResponseWrapper<>(
                      newValue.hash(),
                      null,
                      12
                  ),
                  1L
              ),
              forwarded.get(1).record());
      }
  
      @Test
      public void shouldDeleteKeyNoPropagateV0() {
          final MockProcessorContext<String, SubscriptionResponseWrapper<String>> context = new MockProcessorContext<>();
          processor.init(context);
  
          final SubscriptionWrapper<String> newValue = new SubscriptionWrapper<>(
              new long[]{1L},
              Instruction.DELETE_KEY_NO_PROPAGATE,
              "pk1",
              SubscriptionWrapper.VERSION_0,
              null);
          final Record<CombinedKey<String, String>, Change<ValueAndTimestamp<SubscriptionWrapper<String>>>> record =
              new Record<>(
                  new CombinedKey<>("fk1", "pk1"),
                  new Change<>(ValueAndTimestamp.make(newValue, 1L), null),
                  1L
              );
          processor.process(record);
          final List<CapturedForward<? extends String, ? extends SubscriptionResponseWrapper<String>>> forwarded = context.forwarded();
          assertEquals(0, forwarded.size());
      }
  
      @Test
      public void shouldDeleteKeyNoPropagateV1() {
          final MockProcessorContext<String, SubscriptionResponseWrapper<String>> context = new MockProcessorContext<>();
          processor.init(context);
  
          final SubscriptionWrapper<String> newValue = new SubscriptionWrapper<>(
              new long[]{1L},
              Instruction.DELETE_KEY_NO_PROPAGATE,
              "pk1",
              SubscriptionWrapper.VERSION_1,
              12);
          final Record<CombinedKey<String, String>, Change<ValueAndTimestamp<SubscriptionWrapper<String>>>> record =
              new Record<>(new CombinedKey<>("fk1", "pk1"),
                           new Change<>(ValueAndTimestamp.make(newValue, 1L), null),
                          1L
              );
          processor.process(record);
          final List<CapturedForward<? extends String, ? extends SubscriptionResponseWrapper<String>>> forwarded = context.forwarded();
          assertEquals(0, forwarded.size());
      }
  
      private KTableValueGetterSupplier<String, String> valueGetterSupplier(final Map<String, ValueAndTimestamp<String>> map) {
          final KTableValueGetter<String, String> valueGetter = new KTableValueGetter<String, String>() {
  
              @Override
              public ValueAndTimestamp<String> get(final String key) {
                  return map.get(key);
              }
  
              @Override
              public void init(final ProcessorContext context) {
  
              }
  
              @Override
              public boolean isVersioned() {
                  return false;
              }
          };
          return new KTableValueGetterSupplier<String, String>() {
              @Override
              public KTableValueGetter<String, String> get() {
                  return valueGetter;
              }
  
              @Override
              public String[] storeNames() {
                  return new String[0];
              }
          };
      }
  
      private Processor<CombinedKey<String, String>,
                        Change<ValueAndTimestamp<SubscriptionWrapper<String>>>,
                        String,
                        SubscriptionResponseWrapper<String>> processor(final KTableValueGetterSupplier<String, String> valueGetterSupplier) {
          final SubscriptionJoinProcessorSupplier<String, String, String> supplier =
              new SubscriptionJoinProcessorSupplier<>(valueGetterSupplier);
          return supplier.get();
      }
  }
  