@Test public void testConfigurePipelineSchemaWithMissingInputNameField() { Schema inputSchemaString = Schema.recordOf( "event", Schema.Field.of("CLF", Schema.of(Schema.Type.STRING)) ); MockPipelineConfigurer mockConfigurer = new MockPipelineConfigurer(inputSchemaString, Collections.emptyMap()); S3_TRANSFORM.configurePipeline(mockConfigurer); FailureCollector collector = mockConfigurer.getStageConfigurer().getFailureCollector(); Assert.assertEquals(1, collector.getValidationFailures().size()); Assert.assertEquals(1, collector.getValidationFailures().get(0).getCauses().size()); Cause expectedCause = new Cause(); expectedCause.addAttribute(CauseAttributes.STAGE_CONFIG, LogParserTransform.LogParserConfig.INPUT_NAME); Assert.assertEquals(expectedCause, collector.getValidationFailures().get(0).getCauses().get(0)); }
@Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); if (!S3_LOG.equals(config.logFormat) && !CLF_LOG.equals(config.logFormat) && !CLOUDFRONT_LOG.equals(config.logFormat)) { LOG.error("Log format not currently supported."); collector.addFailure(String.format("Format '%s' is not supported.", config.logFormat), String.format("Supported formats are: %s, %s, and %s.", S3_LOG, CLF_LOG, CLOUDFRONT_LOG)) .withConfigProperty(LogParserConfig.LOG_FORMAT); } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { if (!inputSchema.getType().equals(Schema.Type.RECORD)) { collector.addFailure("Input schema must be of type record.", null); } else if (inputSchema.getField(config.inputName) == null) { collector.addFailure( String.format("Field '%s' must be present in the input schema.", config.inputName), null) .withConfigProperty(LogParserConfig.INPUT_NAME); } else { Schema schema = inputSchema.getField(config.inputName).getSchema(); if (schema.isNullable()) { schema = schema.getNonNullable(); } validateInputSchemaType(schema, config.inputName, collector); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(LOG_SCHEMA); }
LogParserTransform extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); if (!S3_LOG.equals(config.logFormat) && !CLF_LOG.equals(config.logFormat) && !CLOUDFRONT_LOG.equals(config.logFormat)) { LOG.error("Log format not currently supported."); collector.addFailure(String.format("Format '%s' is not supported.", config.logFormat), String.format("Supported formats are: %s, %s, and %s.", S3_LOG, CLF_LOG, CLOUDFRONT_LOG)) .withConfigProperty(LogParserConfig.LOG_FORMAT); } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { if (!inputSchema.getType().equals(Schema.Type.RECORD)) { collector.addFailure("Input schema must be of type record.", null); } else if (inputSchema.getField(config.inputName) == null) { collector.addFailure( String.format("Field '%s' must be present in the input schema.", config.inputName), null) .withConfigProperty(LogParserConfig.INPUT_NAME); } else { Schema schema = inputSchema.getField(config.inputName).getSchema(); if (schema.isNullable()) { schema = schema.getNonNullable(); } validateInputSchemaType(schema, config.inputName, collector); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(LOG_SCHEMA); } }
LogParserTransform extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); if (!S3_LOG.equals(config.logFormat) && !CLF_LOG.equals(config.logFormat) && !CLOUDFRONT_LOG.equals(config.logFormat)) { LOG.error("Log format not currently supported."); collector.addFailure(String.format("Format '%s' is not supported.", config.logFormat), String.format("Supported formats are: %s, %s, and %s.", S3_LOG, CLF_LOG, CLOUDFRONT_LOG)) .withConfigProperty(LogParserConfig.LOG_FORMAT); } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { if (!inputSchema.getType().equals(Schema.Type.RECORD)) { collector.addFailure("Input schema must be of type record.", null); } else if (inputSchema.getField(config.inputName) == null) { collector.addFailure( String.format("Field '%s' must be present in the input schema.", config.inputName), null) .withConfigProperty(LogParserConfig.INPUT_NAME); } else { Schema schema = inputSchema.getField(config.inputName).getSchema(); if (schema.isNullable()) { schema = schema.getNonNullable(); } validateInputSchemaType(schema, config.inputName, collector); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(LOG_SCHEMA); } LogParserTransform(LogParserConfig config); }
LogParserTransform extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); if (!S3_LOG.equals(config.logFormat) && !CLF_LOG.equals(config.logFormat) && !CLOUDFRONT_LOG.equals(config.logFormat)) { LOG.error("Log format not currently supported."); collector.addFailure(String.format("Format '%s' is not supported.", config.logFormat), String.format("Supported formats are: %s, %s, and %s.", S3_LOG, CLF_LOG, CLOUDFRONT_LOG)) .withConfigProperty(LogParserConfig.LOG_FORMAT); } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { if (!inputSchema.getType().equals(Schema.Type.RECORD)) { collector.addFailure("Input schema must be of type record.", null); } else if (inputSchema.getField(config.inputName) == null) { collector.addFailure( String.format("Field '%s' must be present in the input schema.", config.inputName), null) .withConfigProperty(LogParserConfig.INPUT_NAME); } else { Schema schema = inputSchema.getField(config.inputName).getSchema(); if (schema.isNullable()) { schema = schema.getNonNullable(); } validateInputSchemaType(schema, config.inputName, collector); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(LOG_SCHEMA); } LogParserTransform(LogParserConfig config); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void prepareRun(StageSubmitterContext context); @Override void transform(StructuredRecord input, Emitter<StructuredRecord> emitter); }
LogParserTransform extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); if (!S3_LOG.equals(config.logFormat) && !CLF_LOG.equals(config.logFormat) && !CLOUDFRONT_LOG.equals(config.logFormat)) { LOG.error("Log format not currently supported."); collector.addFailure(String.format("Format '%s' is not supported.", config.logFormat), String.format("Supported formats are: %s, %s, and %s.", S3_LOG, CLF_LOG, CLOUDFRONT_LOG)) .withConfigProperty(LogParserConfig.LOG_FORMAT); } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { if (!inputSchema.getType().equals(Schema.Type.RECORD)) { collector.addFailure("Input schema must be of type record.", null); } else if (inputSchema.getField(config.inputName) == null) { collector.addFailure( String.format("Field '%s' must be present in the input schema.", config.inputName), null) .withConfigProperty(LogParserConfig.INPUT_NAME); } else { Schema schema = inputSchema.getField(config.inputName).getSchema(); if (schema.isNullable()) { schema = schema.getNonNullable(); } validateInputSchemaType(schema, config.inputName, collector); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(LOG_SCHEMA); } LogParserTransform(LogParserConfig config); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void prepareRun(StageSubmitterContext context); @Override void transform(StructuredRecord input, Emitter<StructuredRecord> emitter);  }