Added implementation to parse substring as timestamp field

Author: AkshatJindal1

README.md

@@ -456,7 +456,9 @@ Uses [Jackson](https://github.com/FasterXML/jackson) to look for the records in
            Implementation based on based on a `DateTimeFormatter`
 >     *   `com.github.castorm.kafka.connect.http.response.timestamp.NattyTimestampParser`
            Implementation based on [Natty](http://natty.joestelmach.com/) parser
-> 
+>     *   `com.github.castorm.kafka.connect.http.response.timestamp.RegexTimestampParser`
+          Implementation that extracts substring from timestamp column and parse it 
+
 > ##### `http.response.record.timestamp.parser.pattern`
 > When using `DateTimeFormatterTimestampParser`, a custom pattern can be specified 
 > *   Type: `String`
@@ -467,6 +469,16 @@ Uses [Jackson](https://github.com/FasterXML/jackson) to look for the records in
   identifiers
 > *   Type: `String`
 > *   Default: `UTC`
+>
+> ##### `http.response.record.timestamp.parser.regex`
+> When using `RegexTimestampParser`, a custom regex pattern can be specified 
+> *   Type: `String`
+> *   Default: `.*`
+>
+##### `http.response.record.timestamp.parser.regex.delegate`
+> When using `RegexTimestampParser`, a delegate class to parse timestamp 
+> *   Type: `Class`
+> *   Default: `DateTimeFormatterTimestampParser`
 
 ---
 <a name="mapper"/>

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/response/timestamp/RegexTimestampParser.java

@@ -0,0 +1,56 @@
+package com.github.castorm.kafka.connect.http.response.timestamp;
+
+/*-
+ * #%L
+ * Kafka Connect HTTP Plugin
+ * %%
+ * Copyright (C) 2020 CastorM
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+import com.github.castorm.kafka.connect.http.response.timestamp.spi.TimestampParser;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import java.time.Instant;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+@RequiredArgsConstructor
+@Slf4j
+public class RegexTimestampParser implements TimestampParser {
+
+  private final Function<Map<String, ?>, RegexTimestampParserConfig> configFactory;
+  private Pattern pattern;
+  private TimestampParser delegate;
+
+  @Override
+  public Instant parse(String timestamp) {
+    Matcher matcher = pattern.matcher(timestamp);
+    String extractedTimestamp;
+    matcher.find();
+    extractedTimestamp = matcher.group(1);
+    return delegate.parse(extractedTimestamp);
+  }
+
+  @Override
+  public void configure(Map<String, ?> settings) {
+    RegexTimestampParserConfig config = configFactory.apply(settings);
+    pattern = Pattern.compile(config.getTimestampRegex());
+    delegate = config.getDelegateParser();
+  }
+}

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/response/timestamp/RegexTimestampParserConfig.java

@@ -0,0 +1,54 @@
+package com.github.castorm.kafka.connect.http.response.timestamp;
+
+/*-
+ * #%L
+ * kafka-connect-http
+ * %%
+ * Copyright (C) 2020 CastorM
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+import com.github.castorm.kafka.connect.http.response.timestamp.spi.TimestampParser;
+import lombok.Getter;
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+
+import java.util.Map;
+
+import static org.apache.kafka.common.config.ConfigDef.Importance.HIGH;
+import static org.apache.kafka.common.config.ConfigDef.Importance.LOW;
+import static org.apache.kafka.common.config.ConfigDef.Type.CLASS;
+import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
+
+@Getter
+public class RegexTimestampParserConfig extends AbstractConfig {
+  private static final String ITEM_TIMESTAMP_REGEX = "http.response.record.timestamp.parser.regex";
+  private static final String PARSER_DELEGATE = "http.response.record.timestamp.parser.regex.delegate";
+
+  private final String timestampRegex;
+  private final TimestampParser delegateParser;
+
+  RegexTimestampParserConfig(Map<String, ?> originals) {
+    super(config(), originals);
+    timestampRegex = getString(ITEM_TIMESTAMP_REGEX);
+    delegateParser = getConfiguredInstance(PARSER_DELEGATE, TimestampParser.class);
+  }
+
+  public static ConfigDef config() {
+    return new ConfigDef()
+        .define(ITEM_TIMESTAMP_REGEX, STRING, ".*", LOW, "Timestamp regex pattern in case the timestamp value is wrapped around some other text which is not in well defined format")
+        .define(PARSER_DELEGATE, CLASS, DateTimeFormatterTimestampParser.class, HIGH, "Timestamp Parser Delegate Class");
+  }
+}

kafka-connect-http/src/test/java/com/github/castorm/kafka/connect/http/response/timestamp/RegexTimestampParserConfigTest.java

@@ -0,0 +1,55 @@
+package com.github.castorm.kafka.connect.http.response.timestamp;
+
+/*-
+ * #%L
+ * Kafka Connect HTTP
+ * %%
+ * Copyright (C) 2020 Cástor Rodríguez
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static com.github.castorm.kafka.connect.http.response.timestamp.RegexTimestampParserConfigTest.Fixture.config;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class RegexTimestampParserConfigTest {
+
+  @Test
+  void whenItemTimestampParserDelegateClassConfigured_thenInitialized() {
+    assertThat(config(ImmutableMap.of("http.response.record.timestamp.parser.regex.delegate", "com.github.castorm.kafka.connect.http.response.timestamp.EpochMillisTimestampParser")).getDelegateParser())
+        .isInstanceOf(EpochMillisTimestampParser.class);
+    assertThat(config(ImmutableMap.of("http.response.record.timestamp.parser.regex.delegate", "com.github.castorm.kafka.connect.http.response.timestamp.EpochMillisTimestampParser")).getTimestampRegex())
+        .isEqualTo(".*");
+  }
+
+  @Test
+  void whenItemTimestampParserRegexConfigured_thenInitialized() {
+    assertThat(config(ImmutableMap.of("http.response.record.timestamp.parser.regex", "(?:\\/Date\\()(.*?)(?:\\+0000\\)\\/)")).getDelegateParser())
+        .isInstanceOf(DateTimeFormatterTimestampParser.class);
+    assertThat(config(ImmutableMap.of("http.response.record.timestamp.parser.regex", "(?:\\/Date\\()(.*?)(?:\\+0000\\)\\/)")).getTimestampRegex())
+        .isEqualTo("(?:\\/Date\\()(.*?)(?:\\+0000\\)\\/)");
+  }
+
+
+  interface Fixture {
+    static RegexTimestampParserConfig config(Map<String, String> settings) {
+      return new RegexTimestampParserConfig(settings);
+    }
+  }
+}

kafka-connect-http/src/test/java/com/github/castorm/kafka/connect/http/response/timestamp/RegexTimestampParserTest.java

@@ -0,0 +1,83 @@
+package com.github.castorm.kafka.connect.http.response.timestamp;
+
+/*-
+ * #%L
+ * Kafka Connect HTTP
+ * %%
+ * Copyright (C) 2020 Cástor Rodríguez
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+import com.github.castorm.kafka.connect.http.response.timestamp.spi.TimestampParser;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static com.github.castorm.kafka.connect.http.response.timestamp.RegexTimestampParserTest.Fixture.regex;
+import static java.time.Instant.ofEpochMilli;
+import static java.util.Collections.emptyMap;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.then;
+
+@ExtendWith(MockitoExtension.class)
+public class RegexTimestampParserTest {
+
+  RegexTimestampParser parser;
+
+  @Mock
+  TimestampParser delegate;
+
+  @Mock
+  RegexTimestampParserConfig config;
+
+  @BeforeEach
+  void setUp() {
+
+    parser = new RegexTimestampParser(__ -> config);
+    given(config.getDelegateParser()).willReturn(delegate);
+    given(config.getTimestampRegex()).willReturn(regex);
+    parser.configure(emptyMap());
+  }
+
+  @Test
+  void givenLongFormatter_whenParse_thenDelegated() {
+
+    given(delegate.parse("123456789")).willReturn(ofEpochMilli(123456789L));
+
+    assertThat(parser.parse("Date123456789")).isEqualTo(ofEpochMilli(123456789L));
+  }
+
+  @Test
+  void givenFormatter_whenParse_thenDelegated() {
+
+    parser.parse("Date2011-12-03T10:15:30+01");
+    then(delegate).should().parse("2011-12-03T10:15:30+01");
+  }
+
+  @Test
+  void givenNotNumber_whenParse_thenReturnedFromDelegate() {
+
+    given(delegate.parse("2011-12-03T10:15:30+01")).willReturn(ofEpochMilli(123));
+
+    assertThat(parser.parse("Date2011-12-03T10:15:30+01")).isEqualTo(ofEpochMilli(123));
+  }
+
+  interface Fixture {
+    String regex = "(?:Date)(.*)";
+  }
+}