feat(static): add forEach() to KsqlStruct (MINOR) (#3320)

Iterating across the values and their schemas is a common pattern, so lets add a method to facilitate.

ksql-common/src/main/java/io/confluent/ksql/types/KsqlStruct.java

@@ -25,6 +25,7 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.function.BiConsumer;
 
 /**
  * Instance of {@link io.confluent.ksql.schema.ksql.types.SqlStruct}.
@@ -55,6 +56,14 @@ public List<Optional<?>> values() {
     return values;
   }
 
+  public void forEach(final BiConsumer<? super Field, ? super Optional<?>> consumer) {
+    for (int idx = 0; idx < values.size(); idx++) {
+      final Field field = schema.fields().get(idx);
+      final Optional<?> value = values.get(idx);
+      consumer.accept(field, value);
+    }
+  }
+
   @Override
   public boolean equals(final Object o) {
     if (this == o) {

ksql-common/src/test/java/io/confluent/ksql/types/KsqlStructTest.java

@@ -17,6 +17,7 @@
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.contains;
+import static org.mockito.Mockito.inOrder;
 
 import io.confluent.ksql.schema.ksql.DataException;
 import io.confluent.ksql.schema.ksql.Field;
@@ -25,10 +26,16 @@
 import io.confluent.ksql.schema.ksql.types.SqlTypes;
 import io.confluent.ksql.util.KsqlException;
 import java.util.Optional;
+import java.util.function.BiConsumer;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
 
+@RunWith(MockitoJUnitRunner.class)
 public class KsqlStructTest {
 
   private static final SqlStruct SCHEMA = SqlTypes.struct()
@@ -39,6 +46,9 @@ public class KsqlStructTest {
   @Rule
   public final ExpectedException expectedException = ExpectedException.none();
 
+  @Mock
+  private BiConsumer<? super Field, ? super Optional<?>> consumer;
+
   @Test
   public void shouldHandleExplicitNulls() {
     // When:
@@ -94,4 +104,27 @@ public void shouldBuildStruct() {
     // Then:
     assertThat(struct.values(), contains(Optional.of(10L), Optional.of(true)));
   }
+
+  @Test
+  public void shouldVisitFieldsInOrder() {
+    // Given:
+    final KsqlStruct struct = KsqlStruct.builder(SCHEMA)
+        .set(FieldName.of("f0"), Optional.of(10L))
+        .set(FieldName.of("s1", "v1"), Optional.of(true))
+        .build();
+
+    // When:
+    struct.forEach(consumer);
+
+    // Then:
+    final InOrder inOrder = inOrder(consumer);
+    inOrder.verify(consumer).accept(
+        struct.schema().fields().get(0),
+        struct.values().get(0)
+    );
+    inOrder.verify(consumer).accept(
+        struct.schema().fields().get(1),
+        struct.values().get(1)
+    );
+  }
 }