### Question:
TimestampExtractionPolicyFactory { public static TimestampExtractionPolicy create( final Schema schema, final String timestampColumnName, final String timestampFormat) { if (timestampColumnName == null) { return new MetadataTimestampExtractionPolicy(); } final String fieldName = StringUtil.cleanQuotes(timestampColumnName.toUpperCase()); final Field timestampField = SchemaUtil.getFieldByName(schema, fieldName) .orElseThrow(() -> new KsqlException(String.format( "No column with the provided timestamp column name in the " + "WITH clause, %s, exists in the defined schema.", fieldName ))); final Schema.Type timestampFieldType = timestampField.schema().type(); if (timestampFieldType == Schema.Type.STRING) { if (timestampFormat == null) { throw new KsqlException("A String timestamp field has been specified without" + " also specifying the " + DdlConfig.TIMESTAMP_FORMAT_PROPERTY.toLowerCase()); } return new StringTimestampExtractionPolicy( fieldName, StringUtil.cleanQuotes(timestampFormat)); } if (timestampFieldType == Schema.Type.INT64) { return new LongColumnTimestampExtractionPolicy(fieldName); } throw new KsqlException( "Timestamp column, " + timestampColumnName + ", should be LONG(INT64)" + " or a String with a " + DdlConfig.TIMESTAMP_FORMAT_PROPERTY.toLowerCase() + " specified"); }  static TimestampExtractionPolicy create(
      final Schema schema,
      final String timestampColumnName,
      final String timestampFormat);  }### Answer:
@Test public void shouldCreateMetadataPolicyWhenTimestampFieldNotProvided() { assertThat(TimestampExtractionPolicyFactory.create(schemaBuilder.build(), null, null), instanceOf(MetadataTimestampExtractionPolicy.class)); }
@Test public void shouldCreateLongTimestampPolicyWhenTimestampFieldIsOfTypeLong() { final String timestamp = "timestamp"; final Schema schema = schemaBuilder .field(timestamp.toUpperCase(), Schema.INT64_SCHEMA) .build(); TimestampExtractionPolicy extractionPolicy = TimestampExtractionPolicyFactory.create(schema, timestamp, null); assertThat(extractionPolicy, instanceOf(LongColumnTimestampExtractionPolicy.class)); assertThat(extractionPolicy.timestampField(), equalTo(timestamp.toUpperCase())); }
@Test(expected = KsqlException.class) public void shouldFailIfCantFindTimestampField() { TimestampExtractionPolicyFactory.create(schemaBuilder.build(), "whateva", null); }
@Test public void shouldCreateStringTimestampPolicyWhenTimestampFieldIsStringTypeAndFormatProvided() { final String field = "my_string_field"; final Schema schema = schemaBuilder .field(field.toUpperCase(), Schema.STRING_SCHEMA) .build(); TimestampExtractionPolicy extractionPolicy = TimestampExtractionPolicyFactory.create(schema, field, "yyyy-MM-DD"); assertThat(extractionPolicy, instanceOf(StringTimestampExtractionPolicy.class)); assertThat(extractionPolicy.timestampField(), equalTo(field.toUpperCase())); }
@Test(expected = KsqlException.class) public void shouldFailIfStringTimestampTypeAndFormatNotSupplied() { final String field = "my_string_field"; final Schema schema = schemaBuilder .field(field.toUpperCase(), Schema.STRING_SCHEMA) .build(); TimestampExtractionPolicyFactory.create(schema, field, null); }
@Test public void shouldSupportFieldsWithQuotedStrings() { final String field = "my_string_field"; final Schema schema = schemaBuilder .field(field.toUpperCase(), Schema.STRING_SCHEMA) .build(); TimestampExtractionPolicy extractionPolicy = TimestampExtractionPolicyFactory.create(schema, "'"+ field+ "'", "'yyyy-MM-DD'"); assertThat(extractionPolicy, instanceOf(StringTimestampExtractionPolicy.class)); assertThat(extractionPolicy.timestampField(), equalTo(field.toUpperCase())); }
@Test(expected = KsqlException.class) public void shouldThrowIfTimestampFieldTypeIsNotLongOrString() { final String field = "blah"; final Schema schema = schemaBuilder .field(field.toUpperCase(), Schema.FLOAT64_SCHEMA) .build(); TimestampExtractionPolicyFactory.create(schema, "'"+ field+ "'", null); }