add simple extractor from a field

Author: Mike Skells

commons/src/main/java/io/aiven/kafka/connect/common/config/TimestampSource.java

@@ -20,6 +20,9 @@
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 
+import org.apache.kafka.connect.data.Field;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.header.Header;
 import org.apache.kafka.connect.sink.SinkRecord;
 
@@ -148,11 +151,11 @@ public ZonedDateTime time(final SinkRecord record) {
     }
 
     final class DataTimestampSource extends AbstractTimestampSource {
-        private final String field;
+        private final String fieldName;
 
         DataTimestampSource(final String field, final ZoneId zoneId) {
             super(zoneId, Type.DATA);
-            this.field = field;
+            this.fieldName = field;
         }
 
         @Override
@@ -162,7 +165,12 @@ public ZonedDateTime time(final SinkRecord record) {
         }
 
         private Object extract(Object value) {
-            //TODO
+            if (value instanceof Struct) {
+                Struct data = (Struct) value;
+                Schema schema = data.schema();
+                Field field = schema.field(fieldName);
+                return field == null ? null : data.get(field);
+            }
             return null;
         }
     }