Skip to content

Commit

Permalink
Explain Avro's unnecessary EOFException (apache#4098) (apache#4100)
Browse files Browse the repository at this point in the history
* Explain Avro's unnecessary EOFException (apache#4098)

* add jira link into log message
  • Loading branch information
asdf2014 authored and himanshug committed Mar 24, 2017
1 parent 2cbc476 commit 23f77eb
Show file tree
Hide file tree
Showing 3 changed files with 28 additions and 12 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -25,18 +25,17 @@
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;

import io.druid.guice.annotations.Json;
import io.druid.java.util.common.logger.Logger;
import io.druid.java.util.common.parsers.ParseException;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.util.ByteBufferInputStream;

import java.io.EOFException;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.Map;
Expand All @@ -45,7 +44,7 @@
*/
public class InlineSchemaAvroBytesDecoder implements AvroBytesDecoder
{
private static final Logger logger = new Logger(InlineSchemaAvroBytesDecoder.class);
private static final Logger LOGGER = new Logger(InlineSchemaAvroBytesDecoder.class);

private final Schema schemaObj;
private final Map<String, Object> schema;
Expand All @@ -59,9 +58,9 @@ public InlineSchemaAvroBytesDecoder(
Preconditions.checkArgument(schema != null, "schema must be provided");

this.schema = schema;
String schemaStr = mapper.writeValueAsString(schema);;
String schemaStr = mapper.writeValueAsString(schema);

logger.info("Schema string [%s]", schemaStr);
LOGGER.info("Schema string [%s]", schemaStr);
schemaObj = new Schema.Parser().parse(schemaStr);
}

Expand All @@ -87,6 +86,12 @@ public GenericRecord parse(ByteBuffer bytes)
try {
return reader.read(null, DecoderFactory.get().binaryDecoder(inputStream, null));
}
catch (EOFException eof) {
// waiting for avro v1.9.0 (#AVRO-813)
throw new ParseException(
eof, "Avro's unnecessary EOFException, detail: [%s]", "https://issues.apache.org/jira/browse/AVRO-813"
);
}
catch (Exception e) {
throw new ParseException(e, "Fail to decode avro message!");
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,18 +25,17 @@
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;

import io.druid.guice.annotations.Json;
import io.druid.java.util.common.logger.Logger;
import io.druid.java.util.common.parsers.ParseException;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.util.ByteBufferInputStream;

import java.io.EOFException;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashMap;
Expand All @@ -46,7 +45,7 @@
*/
public class InlineSchemasAvroBytesDecoder implements AvroBytesDecoder
{
private static final Logger logger = new Logger(InlineSchemasAvroBytesDecoder.class);
private static final Logger LOGGER = new Logger(InlineSchemasAvroBytesDecoder.class);

private static final byte V1 = 0x1;

Expand All @@ -72,9 +71,10 @@ public InlineSchemasAvroBytesDecoder(
int id = Integer.parseInt(e.getKey());

Map<String, Object> schema = e.getValue();
String schemaStr = mapper.writeValueAsString(schema);;
String schemaStr = mapper.writeValueAsString(schema);
;

logger.info("Schema string [%s] = [%s]", id, schemaStr);
LOGGER.info("Schema string [%s] = [%s]", id, schemaStr);
schemaObjs.put(id, new Schema.Parser().parse(schemaStr));
}
}
Expand Down Expand Up @@ -122,6 +122,12 @@ public GenericRecord parse(ByteBuffer bytes)

return reader.read(null, DecoderFactory.get().binaryDecoder(inputStream, null));
}
catch (EOFException eof) {
// waiting for avro v1.9.0 (#AVRO-813)
throw new ParseException(
eof, "Avro's unnecessary EOFException, detail: [%s]", "https://issues.apache.org/jira/browse/AVRO-813"
);
}
catch (Exception e) {
throw new ParseException(e, "Fail to decode avro message with schemaId [%s].", schemaId);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,9 @@

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;

import io.druid.data.input.schemarepo.SubjectAndIdConverter;
import io.druid.java.util.common.Pair;
import io.druid.java.util.common.parsers.ParseException;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
Expand All @@ -35,6 +33,7 @@
import org.schemarepo.api.TypedSchemaRepository;
import org.schemarepo.api.converter.AvroSchemaConverter;

import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collections;
Expand Down Expand Up @@ -83,6 +82,12 @@ public GenericRecord parse(ByteBuffer bytes)
try {
return reader.read(null, DecoderFactory.get().binaryDecoder(inputStream, null));
}
catch (EOFException eof) {
// waiting for avro v1.9.0 (#AVRO-813)
throw new ParseException(
eof, "Avro's unnecessary EOFException, detail: [%s]", "https://issues.apache.org/jira/browse/AVRO-813"
);
}
catch (IOException e) {
throw new ParseException(e, "Fail to decode avro message!");
}
Expand Down

0 comments on commit 23f77eb

Please sign in to comment.