Fail with appropriate message instead of aborting with NPE when field specified in schema.keyfield doesn't exist in the schema (#83)

Author: yashmayya

README.md

@@ -141,7 +141,7 @@ For an example of using the the Protobuf converter with kafka-connect-datagen, s
 ## Use a bundled schema specification
 
 There are a few quickstart schema specifications bundled with `kafka-connect-datagen`, and they are listed in this [directory](https://github.com/confluentinc/kafka-connect-datagen/tree/master/src/main/resources).
-To use one of these bundled schema, refer to [this mapping](https://github.com/confluentinc/kafka-connect-datagen/blob/master/src/main/java/io/confluent/kafka/connect/datagen/DatagenTask.java#L66-L73) and in the configuration file, set the parameter `quickstart` to the associated name.
+To use one of these bundled schema, refer to [this mapping](https://github.com/confluentinc/kafka-connect-datagen/blob/master/src/main/java/io/confluent/kafka/connect/datagen/DatagenTask.java#L75-L86) and in the configuration file, set the parameter `quickstart` to the associated name.
 For example:
 
 ```bash

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

@@ -31,6 +31,7 @@
 import io.confluent.connect.avro.AvroData;
 import org.apache.avro.generic.GenericData.Record;
 import org.apache.avro.generic.GenericRecord;
+import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.connect.data.SchemaAndValue;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.header.ConnectHeaders;
@@ -182,6 +183,12 @@ public void start(Map<String, String> props) {
     }
 
     avroSchema = generator.schema();
+
+    if (!schemaKeyField.isEmpty() && avroSchema.getField(schemaKeyField) == null) {
+      throw new ConfigException(DatagenConnectorConfig.SCHEMA_KEYFIELD_CONF, schemaKeyField,
+              "Schema does not contain the specified field");
+    }
+
     avroData = new AvroData(1);
     ksqlSchema = avroData.toConnectSchema(avroSchema);
   }

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

@@ -35,6 +35,7 @@
 
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.Struct;
@@ -125,6 +126,8 @@ public void shouldGenerateFilesForPageviewsQuickstart() throws Exception {
   public void shouldGenerateFilesForStockTradesQuickstart() throws Exception {
     generateAndValidateRecordsFor(DatagenTask.Quickstart.STOCK_TRADES);
   }
+
+  @Test
   public void shouldGenerateFilesForProductQuickstart() throws Exception {
     generateAndValidateRecordsFor(Quickstart.PRODUCT);
   }
@@ -201,6 +204,14 @@ public void shouldFailToGenerateMoreRecordsThanSpecified() throws Exception {
     }
   }
 
+  @Test(expected = ConfigException.class)
+  public void shouldFailIfSchemaKeyFieldNotPresentInSchema() throws Exception {
+    config.put(DatagenConnectorConfig.QUICKSTART_CONF, DatagenTask.Quickstart.USERS.name());
+    config.put(DatagenConnectorConfig.SCHEMA_KEYFIELD_CONF, "key_does_not_exist");
+    createTask();
+    generateRecords();
+  }
+
   private void generateAndValidateRecordsFor(DatagenTask.Quickstart quickstart) throws Exception {
     createTaskWith(quickstart);
     generateRecords();