Added Unit Tests for timeout and stackoverflow error

Author: garrix-fan

src/main/java/io/confluent/kafka/connect/datagen/DatagenConnectorConfig.java

@@ -49,7 +49,7 @@ public class DatagenConnectorConfig extends AbstractConfig {
   private static final String RANDOM_SEED_DOC = "Numeric seed for generating random data. "
       + "Two connectors started with the same seed will deterministically produce the same data. "
       + "Each task will generate different data than the other tasks in the same connector.";
-  private static final String GENERATE_TIMEOUT_CONF = "generate.timeout";
+  public static final String GENERATE_TIMEOUT_CONF = "generate.timeout";
   private static final String GENERATE_TIMEOUT_DOC = "Timeout in milliseconds for random message "
       + "generation";
 

src/test/java/io/confluent/kafka/connect/datagen/DatagenTaskTest.java

@@ -44,7 +44,9 @@
 import org.apache.kafka.connect.storage.OffsetStorageReader;
 import org.junit.After;
 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
@@ -61,6 +63,9 @@ public class DatagenTaskTest {
 
   private static final AvroData AVRO_DATA = new AvroData(20);
 
+  @Rule
+  public ExpectedException expectedException = ExpectedException.none();
+
   private Map<String, String> config;
   private DatagenTask task;
   private List<SourceRecord> records;
@@ -223,6 +228,36 @@ public void shouldFailToGenerateMoreRecordsThanSpecified() throws Exception {
     }
   }
 
+  /**
+   * <p>
+   *   This test is non-deterministic. This test might fail in case avro-random-generator library
+   *   updates/changes its underlying implementation.
+   * </p>
+   * @throws Exception
+   */
+  @Test
+  public void shouldFailToGenerateComplexRegexWithTimeout() throws Exception {
+    createTaskWithSchemaAndTimeout("regex_schema_timeout.avro", "regex_key", "100");
+    expectedException.expect(ConnectException.class);
+    expectedException.expectMessage("Record generation timed out");
+    generateRecords();
+  }
+
+  /**
+   * <p>
+   *   This test is non-deterministic. This test might fail in case avro-random-generator library
+   *   updates/changes its underlying implementation.
+   * </p>
+   * @throws Exception
+   */
+  @Test
+  public void shouldFailToGenerateComplexRegexWithStackOverflow() throws Exception {
+    createTaskWithSchemaAndTimeout("regex_schema_stackoverflow.avro", "regex_key", "100");
+    expectedException.expect(ConnectException.class);
+    expectedException.expectMessage("Unable to generate random record");
+    generateRecords();
+  }
+  
   private void generateAndValidateRecordsFor(DatagenTask.Quickstart quickstart) throws Exception {
     createTaskWith(quickstart);
     generateRecords();
@@ -337,6 +372,17 @@ private void createTaskWithSchema(String schemaResourceFilename, String idFieldN
     loadKeyAndValueSchemas(schemaResourceFilename, idFieldName);
   }
 
+  private void createTaskWithSchemaAndTimeout(String schemaResourceFile, String idFieldName,
+    String timeout) {
+    dropSchemaSourceConfigs();
+    config.put(DatagenConnectorConfig.SCHEMA_FILENAME_CONF, schemaResourceFile);
+    config.put(DatagenConnectorConfig.SCHEMA_KEYFIELD_CONF, idFieldName);
+    config.put(DatagenConnectorConfig.GENERATE_TIMEOUT_CONF, timeout);
+    createTask();
+    loadKeyAndValueSchemas(schemaResourceFile, idFieldName);
+
+  }
+
   private void createTaskWithSchemaText(String schemaText, String keyField) {
     dropSchemaSourceConfigs();
     config.put(DatagenConnectorConfig.SCHEMA_STRING_CONF, schemaText);