@Test public void testGetKeySchemaWithoutPrimaryKey() throws IOException { try (InputStream schemaNoPkey = getClass().getClassLoader().getResourceAsStream("dedup-schema/dedup-schema-without-pkey.avsc")) { Schema topicSchema = new Schema.Parser().parse(schemaNoPkey); Schema actualKeySchema = this.runner.getKeySchema(this.job, topicSchema); Assert.assertEquals(actualKeySchema, AvroUtils.removeUncomparableFields(topicSchema).get()); } }
@VisibleForTesting Schema getKeySchema(Job job, Schema topicSchema) throws IOException { Schema keySchema = null; DedupKeyOption dedupKeyOption = getDedupKeyOption(); if (dedupKeyOption == DedupKeyOption.ALL) { LOG.info("Using all attributes in the schema (except Map, Arrar and Enum fields) for compaction"); keySchema = AvroUtils.removeUncomparableFields(topicSchema).get(); } else if (dedupKeyOption == DedupKeyOption.KEY) { LOG.info("Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } else if (keySchemaFileSpecified()) { Path keySchemaFile = getKeySchemaFile(); LOG.info("Using attributes specified in schema file " + keySchemaFile + " for compaction"); try { keySchema = AvroUtils.parseSchemaFromFile(keySchemaFile, this.fs); } catch (IOException e) { LOG.error("Failed to parse avro schema from " + keySchemaFile + ", using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } if (!isKeySchemaValid(keySchema, topicSchema)) { LOG.warn(String.format("Key schema %s is not compatible with record schema %s.", keySchema, topicSchema) + "Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } } else { LOG.info("Property " + COMPACTION_JOB_AVRO_KEY_SCHEMA_LOC + " not provided. Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } return keySchema; }
MRCompactorAvroKeyDedupJobRunner extends MRCompactorJobRunner { @VisibleForTesting Schema getKeySchema(Job job, Schema topicSchema) throws IOException { Schema keySchema = null; DedupKeyOption dedupKeyOption = getDedupKeyOption(); if (dedupKeyOption == DedupKeyOption.ALL) { LOG.info("Using all attributes in the schema (except Map, Arrar and Enum fields) for compaction"); keySchema = AvroUtils.removeUncomparableFields(topicSchema).get(); } else if (dedupKeyOption == DedupKeyOption.KEY) { LOG.info("Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } else if (keySchemaFileSpecified()) { Path keySchemaFile = getKeySchemaFile(); LOG.info("Using attributes specified in schema file " + keySchemaFile + " for compaction"); try { keySchema = AvroUtils.parseSchemaFromFile(keySchemaFile, this.fs); } catch (IOException e) { LOG.error("Failed to parse avro schema from " + keySchemaFile + ", using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } if (!isKeySchemaValid(keySchema, topicSchema)) { LOG.warn(String.format("Key schema %s is not compatible with record schema %s.", keySchema, topicSchema) + "Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } } else { LOG.info("Property " + COMPACTION_JOB_AVRO_KEY_SCHEMA_LOC + " not provided. Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } return keySchema; } }
MRCompactorAvroKeyDedupJobRunner extends MRCompactorJobRunner { @VisibleForTesting Schema getKeySchema(Job job, Schema topicSchema) throws IOException { Schema keySchema = null; DedupKeyOption dedupKeyOption = getDedupKeyOption(); if (dedupKeyOption == DedupKeyOption.ALL) { LOG.info("Using all attributes in the schema (except Map, Arrar and Enum fields) for compaction"); keySchema = AvroUtils.removeUncomparableFields(topicSchema).get(); } else if (dedupKeyOption == DedupKeyOption.KEY) { LOG.info("Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } else if (keySchemaFileSpecified()) { Path keySchemaFile = getKeySchemaFile(); LOG.info("Using attributes specified in schema file " + keySchemaFile + " for compaction"); try { keySchema = AvroUtils.parseSchemaFromFile(keySchemaFile, this.fs); } catch (IOException e) { LOG.error("Failed to parse avro schema from " + keySchemaFile + ", using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } if (!isKeySchemaValid(keySchema, topicSchema)) { LOG.warn(String.format("Key schema %s is not compatible with record schema %s.", keySchema, topicSchema) + "Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } } else { LOG.info("Property " + COMPACTION_JOB_AVRO_KEY_SCHEMA_LOC + " not provided. Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } return keySchema; } MRCompactorAvroKeyDedupJobRunner(Dataset dataset, FileSystem fs); }
MRCompactorAvroKeyDedupJobRunner extends MRCompactorJobRunner { @VisibleForTesting Schema getKeySchema(Job job, Schema topicSchema) throws IOException { Schema keySchema = null; DedupKeyOption dedupKeyOption = getDedupKeyOption(); if (dedupKeyOption == DedupKeyOption.ALL) { LOG.info("Using all attributes in the schema (except Map, Arrar and Enum fields) for compaction"); keySchema = AvroUtils.removeUncomparableFields(topicSchema).get(); } else if (dedupKeyOption == DedupKeyOption.KEY) { LOG.info("Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } else if (keySchemaFileSpecified()) { Path keySchemaFile = getKeySchemaFile(); LOG.info("Using attributes specified in schema file " + keySchemaFile + " for compaction"); try { keySchema = AvroUtils.parseSchemaFromFile(keySchemaFile, this.fs); } catch (IOException e) { LOG.error("Failed to parse avro schema from " + keySchemaFile + ", using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } if (!isKeySchemaValid(keySchema, topicSchema)) { LOG.warn(String.format("Key schema %s is not compatible with record schema %s.", keySchema, topicSchema) + "Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } } else { LOG.info("Property " + COMPACTION_JOB_AVRO_KEY_SCHEMA_LOC + " not provided. Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } return keySchema; } MRCompactorAvroKeyDedupJobRunner(Dataset dataset, FileSystem fs); static Schema getKeySchema(Schema topicSchema); static Optional<Schema> getKeySchema(Field field); static Optional<Schema> getKeySchemaFromRecord(Schema record); static boolean isKeySchemaValid(Schema keySchema, Schema topicSchema); static Schema getNewestSchemaFromSource(Job job, FileSystem fs); static Schema getNewestSchemaFromSource(Path sourceDir, FileSystem fs); }
MRCompactorAvroKeyDedupJobRunner extends MRCompactorJobRunner { @VisibleForTesting Schema getKeySchema(Job job, Schema topicSchema) throws IOException { Schema keySchema = null; DedupKeyOption dedupKeyOption = getDedupKeyOption(); if (dedupKeyOption == DedupKeyOption.ALL) { LOG.info("Using all attributes in the schema (except Map, Arrar and Enum fields) for compaction"); keySchema = AvroUtils.removeUncomparableFields(topicSchema).get(); } else if (dedupKeyOption == DedupKeyOption.KEY) { LOG.info("Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } else if (keySchemaFileSpecified()) { Path keySchemaFile = getKeySchemaFile(); LOG.info("Using attributes specified in schema file " + keySchemaFile + " for compaction"); try { keySchema = AvroUtils.parseSchemaFromFile(keySchemaFile, this.fs); } catch (IOException e) { LOG.error("Failed to parse avro schema from " + keySchemaFile + ", using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } if (!isKeySchemaValid(keySchema, topicSchema)) { LOG.warn(String.format("Key schema %s is not compatible with record schema %s.", keySchema, topicSchema) + "Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } } else { LOG.info("Property " + COMPACTION_JOB_AVRO_KEY_SCHEMA_LOC + " not provided. Using key attributes in the schema for compaction"); keySchema = AvroUtils.removeUncomparableFields(getKeySchema(topicSchema)).get(); } return keySchema; } MRCompactorAvroKeyDedupJobRunner(Dataset dataset, FileSystem fs); static Schema getKeySchema(Schema topicSchema); static Optional<Schema> getKeySchema(Field field); static Optional<Schema> getKeySchemaFromRecord(Schema record); static boolean isKeySchemaValid(Schema keySchema, Schema topicSchema); static Schema getNewestSchemaFromSource(Job job, FileSystem fs); static Schema getNewestSchemaFromSource(Path sourceDir, FileSystem fs); static final String COMPACTION_JOB_AVRO_SINGLE_INPUT_SCHEMA; static final String COMPACTION_JOB_AVRO_KEY_SCHEMA_LOC; static final String COMPACTION_JOB_DEDUP_KEY; static final String COMPACTION_JOB_KEY_FIELD_BLACKLIST; static final DedupKeyOption DEFAULT_DEDUP_KEY_OPTION; }