[FLINK-8543] Don't call super.close() in AvroKeyValueSinkWriter

The call to keyValueWriter.close() in AvroKeyValueSinkWriter.close()
will eventually call flush() on the wrapped stream which fails if we
close it before(). Now we call flush ourselves before closing the
KeyValyeWriter, which internally closes the wrapped stream eventually.

flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/AvroKeyValueSinkWriter.java

@@ -150,17 +150,29 @@ private CodecFactory getCompressionCodec(Map<String, String> conf) {
  public void open(FileSystem fs, Path path) throws IOException {
  super.open(fs, path);
 
- CodecFactory compressionCodec = getCompressionCodec(properties);
- Schema keySchema = Schema.parse(properties.get(CONF_OUTPUT_KEY_SCHEMA));
- Schema valueSchema = Schema.parse(properties.get(CONF_OUTPUT_VALUE_SCHEMA));
- keyValueWriter = new AvroKeyValueWriter<K, V>(keySchema, valueSchema, compressionCodec, getStream());
+ try {
+ CodecFactory compressionCodec = getCompressionCodec(properties);
+ Schema keySchema = Schema.parse(properties.get(CONF_OUTPUT_KEY_SCHEMA));
+ Schema valueSchema = Schema.parse(properties.get(CONF_OUTPUT_VALUE_SCHEMA));
+ keyValueWriter = new AvroKeyValueWriter<K, V>(
+ keySchema,
+ valueSchema,
+ compressionCodec,
+ getStream());
+ } finally {
+ if (keyValueWriter == null) {
+ close();
+ }
+ }
  }
 
  @Override
  public void close() throws IOException {
- super.close(); //the order is important since super.close flushes inside
  if (keyValueWriter != null) {
  keyValueWriter.close();
+ } else {
+ // need to make sure we close this if we never created the Key/Value Writer.
+ super.close();
  }
  }