Added Unit Tests for generate.timeout config

Author: garrix-fan

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

@@ -51,7 +51,7 @@ public class DatagenConnectorConfig extends AbstractConfig {
       + "Each task will generate different data than the other tasks in the same connector.";
   public static final String GENERATE_TIMEOUT_CONF = "generate.timeout";
   private static final String GENERATE_TIMEOUT_DOC = "Timeout in milliseconds for random message "
-      + "generation";
+      + "generation. This timeout can be configured for upto 1 minute, i.e 60000ms";
 
   public DatagenConnectorConfig(ConfigDef config, Map<String, String> parsedConfig) {
     super(config, parsedConfig);

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

@@ -220,6 +220,30 @@ public void shouldNotValidateSchemaKeyFieldWhenMultipleSchemaSourcesAreSet() {
     );
   }
 
+  @Test
+  public void shouldAllowSettingGenerateTimeoutInRange() {
+    clearSchemaSources();
+    config.put(DatagenConnectorConfig.GENERATE_TIMEOUT_CONF, "100");
+    Config validated = connector.validate(config);
+    assertThat(validated, hasNoValidationErrorsFor(DatagenConnectorConfig.GENERATE_TIMEOUT_CONF));
+  }
+
+  @Test
+  public void shouldNotAllowSettingGenerateTimeoutNegative() {
+    clearSchemaSources();
+    config.put(DatagenConnectorConfig.GENERATE_TIMEOUT_CONF, "-1");
+    Config validated = connector.validate(config);
+    assertThat(validated, hasValidationError(DatagenConnectorConfig.GENERATE_TIMEOUT_CONF, 1));
+  }
+
+  @Test
+  public void shouldNotAllowSettingGenerateTimeoutOutOfRange() {
+    clearSchemaSources();
+    config.put(DatagenConnectorConfig.GENERATE_TIMEOUT_CONF, "70000");
+    Config validated = connector.validate(config);
+    assertThat(validated, hasValidationError(DatagenConnectorConfig.GENERATE_TIMEOUT_CONF, 1));
+  }
+
   protected void assertTaskConfigs(int maxTasks) {
     List<Map<String, String>> taskConfigs = connector.taskConfigs(maxTasks);
     assertEquals(maxTasks, taskConfigs.size());