AvroKeyValueSinkWriter extends StreamWriterBase<Tuple2<K, V>> implements Writer<Tuple2<K, V>>, InputTypeConfigurable { @Override public Writer<Tuple2<K, V>> duplicate() { return new AvroKeyValueSinkWriter<>(this); } @SuppressWarnings("deprecation")  AvroKeyValueSinkWriter(Map<String, String> properties); protected  AvroKeyValueSinkWriter(AvroKeyValueSinkWriter<K, V> other); @Override @SuppressWarnings("deprecation") void open(FileSystem fs, Path path); @Override void close(); @Override long flush(); @Override void write(Tuple2<K, V> element); @Override void setInputType(TypeInformation<?> type, ExecutionConfig executionConfig); @Override Writer<Tuple2<K, V>> duplicate(); @Override int hashCode(); @Override boolean equals(Object other); static final String CONF_OUTPUT_KEY_SCHEMA; static final String CONF_OUTPUT_VALUE_SCHEMA; static final String CONF_COMPRESS; static final String CONF_COMPRESS_CODEC; static final String CONF_DEFLATE_LEVEL; static final String CONF_XZ_LEVEL; }
@Test public void testDuplicate() { Map<String, String> properties = new HashMap<>(); Schema keySchema = Schema.create(Schema.Type.STRING); Schema valueSchema = Schema.create(Schema.Type.STRING); properties.put(AvroKeyValueSinkWriter.CONF_OUTPUT_KEY_SCHEMA, keySchema.toString()); properties.put(AvroKeyValueSinkWriter.CONF_OUTPUT_VALUE_SCHEMA, valueSchema.toString()); properties.put(AvroKeyValueSinkWriter.CONF_COMPRESS, String.valueOf(true)); properties.put(AvroKeyValueSinkWriter.CONF_COMPRESS_CODEC, DataFileConstants.SNAPPY_CODEC); AvroKeyValueSinkWriter<String, String> writer = new AvroKeyValueSinkWriter(properties); writer.setSyncOnFlush(true); Writer<Tuple2<String, String>> other = writer.duplicate(); assertTrue(writer.equals(other)); writer.setSyncOnFlush(false); assertFalse(writer.equals(other)); }