shiv4289 commented on a change in pull request #6695: [issue 6694][AVRO ENCODE]
Reset cursor if message encode fails.
URL: https://github.com/apache/pulsar/pull/6695#discussion_r407197590
##########
File path:
pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/writer/AvroWriter.java
##########
@@ -47,6 +47,11 @@ public AvroWriter(Schema schema) {
} catch (Exception e) {
throw new SchemaSerializationException(e);
} finally {
+ try {
+ this.encoder.flush();
Review comment:
Hi @sijie thanks for reviewing this promptly.
Below are two easy (or minimal code change) alternate option I see to skip
flushing again. Just cosmetic difference. I prefer 1 by the way.
1 .
```
public synchronized byte[] write(T message) {
byte[] outputBytes = null;
try {
writer.write(message, this.encoder);
outputBytes = this.byteArrayOutputStream.toByteArray();
} catch (IOException e) {
throw new SchemaSerializationException(e);
} finally {
try {
this.encoder.flush();
} catch (IOException ex) {
throw new SchemaSerializationException(e);
}
this.byteArrayOutputStream.reset();
}
return outputBytes;
}
```
And 2.
```
public synchronized byte[] write(T message) {
try {
writer.write(message, this.encoder);
this.encoder.flush();
return this.byteArrayOutputStream.toByteArray();
} catch (IOException e) {
try {
this.encoder.flush();
} catch (IOException ex) {
throw new SchemaSerializationException(e);
}
throw new SchemaSerializationException(e);
} finally {
this.byteArrayOutputStream.reset();
}
}
```
Regarding your suggestions on slack:
> Can flush throw exception?
Yes. Infact any method for the encoder throws an IOException!
> If flush also throws exception, then we might need to know where the
exception is thrown and whether do we need to call flush.
Flush is
[here](https://github.com/apache/avro/blob/9af218d3c060da103868cdc69d7b39fc8d50dcbc/lang/java/avro/src/main/java/org/apache/avro/io/BufferedBinaryEncoder.java#L84)
and I want to call it in case of crash because of [this
line](https://github.com/apache/avro/blob/9af218d3c060da103868cdc69d7b39fc8d50dcbc/lang/java/avro/src/main/java/org/apache/avro/io/BufferedBinaryEncoder.java#L98).
I understand i need a reset to be precise but there is no way to throw the
contents of the stream and reset the cursor to 0.
> Alternatively, we can consider adding a flag for indicating if the flush
succeeded or not.
The success or failure is in effect separated by a null pointer exception
[here](https://github.com/apache/avro/blob/9af218d3c060da103868cdc69d7b39fc8d50dcbc/lang/java/avro/src/main/java/org/apache/avro/specific/SpecificDatumWriter.java#L68).
With that, it is still some sort of a hack to catch a null pointer exception
and then update a flag. Also, that would require changing the avro library by
the way.
On the other hand, to add a flag on the pulsar side in the current class
will actually mean updating it in the catch/finally which again means there is
not much use to adding a flag.
Please let me know if (1) is ok. Happy to know if you have any other
suggestion(s).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
