KAFKA-19096: Added checks for the number of dropped records to foreign key join tests. (#21111)

Follow up to #20605 to improve test
coverage.

Reviewers: Matthias J. Sax <matthias@confluent.io>

Author: Nikita-Shupletsov

streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignTableJoinProcessorSupplierTests.java

@@ -48,6 +48,7 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
 public class ForeignTableJoinProcessorSupplierTests {
 
@@ -113,6 +114,9 @@ public void shouldPropagateRightRecordForEachMatchingPrimaryKey() {
             context.forwarded().get(1).record(),
             is(new Record<>(pk2, new SubscriptionResponseWrapper<>(hash, "new_value", null), 0))
         );
+
+        // test dropped-records sensors
+        assertEquals(0.0, getDroppedRecordsTotalMetric(context));
     }
 
     @Test
@@ -125,6 +129,9 @@ public void shouldPropagateNothingIfNoMatchingPrimaryKey() {
         processor.process(record);
 
         assertThat(context.forwarded(), empty());
+
+        // test dropped-records sensors
+        assertEquals(0.0, getDroppedRecordsTotalMetric(context));
     }
 
     @Test
@@ -145,6 +152,9 @@ public void shouldPropagateTombstoneRightRecordForEachMatchingPrimaryKey() {
             context.forwarded().get(1).record(),
             is(new Record<>(pk2, new SubscriptionResponseWrapper<>(hash, null, null), 0))
         );
+
+        // test dropped-records sensors
+        assertEquals(0.0, getDroppedRecordsTotalMetric(context));
     }
 
     @Test
@@ -162,6 +172,9 @@ public void shouldNotMatchForeignKeysHavingThisFKAsPrefix() {
             context.forwarded().get(0).record(),
             is(new Record<>(pk2, new SubscriptionResponseWrapper<>(hash, "new_value", null), 0))
         );
+
+        // test dropped-records sensors
+        assertEquals(0.0, getDroppedRecordsTotalMetric(context));
     }
 
     @Test
@@ -175,7 +188,7 @@ public void shouldIgnoreRecordWithNullKey() {
         assertThat(context.forwarded(), empty());
 
         // test dropped-records sensors
-        Assertions.assertEquals(1.0, getDroppedRecordsTotalMetric(context));
+        assertEquals(1.0, getDroppedRecordsTotalMetric(context));
         Assertions.assertNotEquals(0.0, getDroppedRecordsRateMetric(context));
     }
 

streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ResponseJoinProcessorSupplierTest.java

@@ -164,6 +164,9 @@ public void shouldForwardWhenHashMatches() {
         final List<MockProcessorContext.CapturedForward<? extends String, ? extends String>> forwarded = context.forwarded();
         assertThat(forwarded.size(), is(1));
         assertThat(forwarded.get(0).record(), is(new Record<>("lhs1", "(lhsValue,rhsValue)", 0)));
+
+        // test dropped-records sensors
+        assertEquals(0.0, getDroppedRecordsTotalMetric(context));
     }
 
     @Test
@@ -190,6 +193,9 @@ public void shouldEmitTombstoneForInnerJoinWhenRightIsNull() {
         final List<MockProcessorContext.CapturedForward<? extends String, ? extends String>> forwarded = context.forwarded();
         assertThat(forwarded.size(), is(1));
         assertThat(forwarded.get(0).record(), is(new Record<>("lhs1", null, 0)));
+
+        // test dropped-records sensors
+        assertEquals(0.0, getDroppedRecordsTotalMetric(context));
     }
 
     @Test
@@ -216,6 +222,9 @@ public void shouldEmitResultForLeftJoinWhenRightIsNull() {
         final List<MockProcessorContext.CapturedForward<? extends String, ? extends String>> forwarded = context.forwarded();
         assertThat(forwarded.size(), is(1));
         assertThat(forwarded.get(0).record(), is(new Record<>("lhs1", "(lhsValue,null)", 0)));
+
+        // test dropped-records sensors
+        assertEquals(0.0, getDroppedRecordsTotalMetric(context));
     }
 
     @Test
@@ -242,6 +251,9 @@ public void shouldEmitTombstoneForLeftJoinWhenRightIsNullAndLeftIsNull() {
         final List<MockProcessorContext.CapturedForward<? extends String, ? extends String>> forwarded = context.forwarded();
         assertThat(forwarded.size(), is(1));
         assertThat(forwarded.get(0).record(), is(new Record<>("lhs1", null, 0)));
+
+        // test dropped-records sensors
+        assertEquals(0.0, getDroppedRecordsTotalMetric(context));
     }
 
     static Object getDroppedRecordsTotalMetric(final InternalProcessorContext<String, ?> context) {