@Test(expected = IllegalArgumentException.class) public void testTableSinkWithMissingValueField() { Schema inputSchema = Schema.recordOf( "purchase", Schema.Field.of("rowKey", Schema.of(Schema.Type.STRING)), Schema.Field.of("userid", Schema.of(Schema.Type.STRING)) ); KVTableSink.KVTableConfig kvTableConfig = new KVTableSink.KVTableConfig("purchases", "rowKey", "user"); KVTableSink kvTableSink = new KVTableSink(kvTableConfig); MockPipelineConfigurer mockPipelineConfigurer = new MockPipelineConfigurer(inputSchema); kvTableSink.configurePipeline(mockPipelineConfigurer); }
@Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { SchemaValidator.validateFieldsArePresentInSchema(inputSchema, kvTableConfig.keyField); SchemaValidator.validateFieldsArePresentInSchema(inputSchema, kvTableConfig.valueField); validateSchemaTypeIsStringOrBytes(inputSchema, kvTableConfig.keyField, false); validateSchemaTypeIsStringOrBytes(inputSchema, kvTableConfig.valueField, true); } }
KVTableSink extends BatchWritableSink<StructuredRecord, byte[], byte[]> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { SchemaValidator.validateFieldsArePresentInSchema(inputSchema, kvTableConfig.keyField); SchemaValidator.validateFieldsArePresentInSchema(inputSchema, kvTableConfig.valueField); validateSchemaTypeIsStringOrBytes(inputSchema, kvTableConfig.keyField, false); validateSchemaTypeIsStringOrBytes(inputSchema, kvTableConfig.valueField, true); } } }
KVTableSink extends BatchWritableSink<StructuredRecord, byte[], byte[]> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { SchemaValidator.validateFieldsArePresentInSchema(inputSchema, kvTableConfig.keyField); SchemaValidator.validateFieldsArePresentInSchema(inputSchema, kvTableConfig.valueField); validateSchemaTypeIsStringOrBytes(inputSchema, kvTableConfig.keyField, false); validateSchemaTypeIsStringOrBytes(inputSchema, kvTableConfig.valueField, true); } } KVTableSink(KVTableConfig kvTableConfig); }
KVTableSink extends BatchWritableSink<StructuredRecord, byte[], byte[]> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { SchemaValidator.validateFieldsArePresentInSchema(inputSchema, kvTableConfig.keyField); SchemaValidator.validateFieldsArePresentInSchema(inputSchema, kvTableConfig.valueField); validateSchemaTypeIsStringOrBytes(inputSchema, kvTableConfig.keyField, false); validateSchemaTypeIsStringOrBytes(inputSchema, kvTableConfig.valueField, true); } } KVTableSink(KVTableConfig kvTableConfig); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void transform(StructuredRecord input, Emitter<KeyValue<byte[], byte[]>> emitter); }
KVTableSink extends BatchWritableSink<StructuredRecord, byte[], byte[]> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { SchemaValidator.validateFieldsArePresentInSchema(inputSchema, kvTableConfig.keyField); SchemaValidator.validateFieldsArePresentInSchema(inputSchema, kvTableConfig.valueField); validateSchemaTypeIsStringOrBytes(inputSchema, kvTableConfig.keyField, false); validateSchemaTypeIsStringOrBytes(inputSchema, kvTableConfig.valueField, true); } } KVTableSink(KVTableConfig kvTableConfig); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void transform(StructuredRecord input, Emitter<KeyValue<byte[], byte[]>> emitter);  }