public Pair<GenericKey, GenericRow> generateRow() {

    final Object generatedObject = generator.generate();

    if (!(generatedObject instanceof GenericRecord)) {
      throw new RuntimeException(String.format(
          "Expected Avro Random Generator to return instance of GenericRecord, found %s instead",
          generatedObject.getClass().getName()
      ));
    }
    final GenericRecord randomAvroMessage = (GenericRecord) generatedObject;

    final GenericRow row = new GenericRow(generator.schema().getFields().size());

    SimpleDateFormat timeformatter = null;

    /*
     * Populate the record entries
     */
    String sessionisationValue = null;
    for (final Schema.Field field : generator.schema().getFields()) {

      final boolean isSession = field.schema().getProp("session") != null;
      final boolean isSessionSiblingIntHash =
          field.schema().getProp("session-sibling-int-hash") != null;
      final String timeFormatFromLong = field.schema().getProp("format_as_time");

      if (isSession) {
        final String currentValue = (String) randomAvroMessage.get(field.name());
        final String newCurrentValue = handleSessionisationOfValue(sessionManager, currentValue);
        sessionisationValue = newCurrentValue;

        row.append(newCurrentValue);
      } else if (isSessionSiblingIntHash && sessionisationValue != null) {

        // super cheeky hack to link int-ids to session-values - if anything fails then we use
        // the 'avro-gen' randomised version
        handleSessionSiblingField(
            randomAvroMessage,
            row,
            sessionisationValue,
            field
        );

      } else if (timeFormatFromLong != null) {
        final Date date = new Date(System.currentTimeMillis());
        if (timeFormatFromLong.equals("unix_long")) {
          row.append(date.getTime());
        } else {
          if (timeformatter == null) {
            timeformatter = new SimpleDateFormat(timeFormatFromLong);
          }
          row.append(timeformatter.format(date));
        }
      } else {
        final Object value = randomAvroMessage.get(field.name());
        if (value instanceof Record) {
          final Field ksqlField = valueSchema.field(field.name());
          final Record record = (Record) value;
          final Object ksqlValue = avroData.toConnectData(record.getSchema(), record).value();
          row.append(DataGenSchemaUtil.getOptionalValue(ksqlField.schema(), ksqlValue));
        } else {
          row.append(value);
        }
      }
    }

    final GenericKey key = GenericKey.genericKey(row.get(keyFieldIndex));

    return Pair.of(key, row);
  }
@Test
  public void shouldGenerateCorrectKey() throws IOException {
    final Generator generator = new Generator(new File("./src/main/resources/pageviews_schema.avro"), new Random());

    final RowGenerator rowGenerator = new RowGenerator(generator, "viewtime", Optional.empty());

    final Pair<GenericKey, GenericRow> rowPair = rowGenerator.generateRow();

    final GenericKey key = rowPair.getLeft();
    final GenericRow value = rowPair.getRight();
    assertThat(key, is(notNullValue()));
    assertThat(key.get(0), is(instanceOf(Long.class)));

    assertThat("must match copy of key in value", key.get(0), is(value.get(0)));
  }