Skip to content

Commit b9e7836

Browse files
MateuszDobrowolskiMateuszDobrowolski
committed
GH-1808 | Reformat code
1 parent 55d1622 commit b9e7836

File tree

9 files changed

+1084
-1094
lines changed

9 files changed

+1084
-1094
lines changed

‎hermes-common/src/main/java/pl/allegro/tech/hermes/common/message/converter/AvroBinaryDecoders.java

+42-41
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,10 @@
11
package pl.allegro.tech.hermes.common.message.converter;
22

3+
import java.io.ByteArrayInputStream;
4+
import java.io.Closeable;
5+
import java.io.InputStream;
6+
import java.util.Collection;
7+
import java.util.Collections;
38
import org.apache.avro.Conversion;
49
import org.apache.avro.Schema;
510
import org.apache.avro.generic.GenericData;
@@ -10,53 +15,49 @@
1015
import org.apache.commons.lang3.exception.ExceptionUtils;
1116
import tech.allegro.schema.json2avro.converter.AvroConversionException;
1217

13-
import java.io.ByteArrayInputStream;
14-
import java.io.Closeable;
15-
import java.io.InputStream;
16-
import java.util.Collection;
17-
import java.util.Collections;
18-
1918
public class AvroBinaryDecoders {
2019

21-
private static ThreadLocal<InputStream> threadLocalEmptyInputStream =
22-
ThreadLocal.withInitial(() -> new ByteArrayInputStream(new byte[0]));
23-
24-
private static ThreadLocal<BinaryDecoder> threadLocalBinaryDecoder =
25-
ThreadLocal.withInitial(
26-
() -> DecoderFactory.get().binaryDecoder(threadLocalEmptyInputStream.get(), null));
27-
28-
static GenericRecord decodeReusingThreadLocalBinaryDecoder(byte[] message, Schema schema) {
29-
return decodeReusingThreadLocalBinaryDecoder(message, schema, Collections.emptyList());
30-
}
31-
32-
static GenericRecord decodeReusingThreadLocalBinaryDecoder(byte[] message, Schema schema, Collection<Conversion<?>> conversions) {
33-
try (FlushableBinaryDecoderHolder holder = new FlushableBinaryDecoderHolder()) {
34-
BinaryDecoder binaryDecoder =
35-
DecoderFactory.get().binaryDecoder(message, holder.getBinaryDecoder());
36-
GenericData genericData = new GenericData();
37-
conversions.forEach(genericData::addLogicalTypeConversion);
38-
return new GenericDatumReader<GenericRecord>(schema, schema, genericData).read(null, binaryDecoder);
39-
} catch (Exception e) {
40-
String reason =
41-
e.getMessage() == null ? ExceptionUtils.getRootCauseMessage(e) : e.getMessage();
42-
throw new AvroConversionException(
43-
String.format(
44-
"Could not deserialize Avro message with provided schema, reason: %s", reason));
45-
}
20+
private static ThreadLocal<InputStream> threadLocalEmptyInputStream =
21+
ThreadLocal.withInitial(() -> new ByteArrayInputStream(new byte[0]));
22+
23+
private static ThreadLocal<BinaryDecoder> threadLocalBinaryDecoder =
24+
ThreadLocal.withInitial(
25+
() -> DecoderFactory.get().binaryDecoder(threadLocalEmptyInputStream.get(), null));
26+
27+
static GenericRecord decodeReusingThreadLocalBinaryDecoder(byte[] message, Schema schema) {
28+
return decodeReusingThreadLocalBinaryDecoder(message, schema, Collections.emptyList());
29+
}
30+
31+
static GenericRecord decodeReusingThreadLocalBinaryDecoder(
32+
byte[] message, Schema schema, Collection<Conversion<?>> conversions) {
33+
try (FlushableBinaryDecoderHolder holder = new FlushableBinaryDecoderHolder()) {
34+
BinaryDecoder binaryDecoder =
35+
DecoderFactory.get().binaryDecoder(message, holder.getBinaryDecoder());
36+
GenericData genericData = new GenericData();
37+
conversions.forEach(genericData::addLogicalTypeConversion);
38+
return new GenericDatumReader<GenericRecord>(schema, schema, genericData)
39+
.read(null, binaryDecoder);
40+
} catch (Exception e) {
41+
String reason =
42+
e.getMessage() == null ? ExceptionUtils.getRootCauseMessage(e) : e.getMessage();
43+
throw new AvroConversionException(
44+
String.format(
45+
"Could not deserialize Avro message with provided schema, reason: %s", reason));
4646
}
47+
}
4748

48-
static class FlushableBinaryDecoderHolder implements Closeable {
49+
static class FlushableBinaryDecoderHolder implements Closeable {
4950

50-
final BinaryDecoder binaryDecoder = threadLocalBinaryDecoder.get();
51+
final BinaryDecoder binaryDecoder = threadLocalBinaryDecoder.get();
5152

52-
BinaryDecoder getBinaryDecoder() {
53-
return binaryDecoder;
54-
}
53+
BinaryDecoder getBinaryDecoder() {
54+
return binaryDecoder;
55+
}
5556

56-
@Override
57-
public void close() {
58-
DecoderFactory.get()
59-
.binaryDecoder(threadLocalEmptyInputStream.get(), threadLocalBinaryDecoder.get());
60-
}
57+
@Override
58+
public void close() {
59+
DecoderFactory.get()
60+
.binaryDecoder(threadLocalEmptyInputStream.get(), threadLocalBinaryDecoder.get());
6161
}
62+
}
6263
}

‎hermes-common/src/main/java/pl/allegro/tech/hermes/common/message/converter/AvroRecordToBytesConverter.java

+17-17
Original file line numberDiff line numberDiff line change
@@ -1,32 +1,32 @@
11
package pl.allegro.tech.hermes.common.message.converter;
22

3+
import java.io.ByteArrayOutputStream;
4+
import java.io.IOException;
5+
import java.util.Collection;
36
import org.apache.avro.Conversion;
47
import org.apache.avro.Schema;
58
import org.apache.avro.generic.GenericDatumWriter;
69
import org.apache.avro.generic.GenericRecord;
710
import org.apache.avro.io.BinaryEncoder;
811
import org.apache.avro.io.EncoderFactory;
912

10-
import java.io.ByteArrayOutputStream;
11-
import java.io.IOException;
12-
import java.util.Collection;
13-
1413
public interface AvroRecordToBytesConverter {
1514

16-
static GenericRecord bytesToRecord(byte[] data, Schema schema) {
17-
return AvroBinaryDecoders.decodeReusingThreadLocalBinaryDecoder(data, schema);
18-
}
15+
static GenericRecord bytesToRecord(byte[] data, Schema schema) {
16+
return AvroBinaryDecoders.decodeReusingThreadLocalBinaryDecoder(data, schema);
17+
}
1918

20-
static GenericRecord bytesToRecord(byte[] data, Schema schema, Collection<Conversion<?>> conversions) {
21-
return AvroBinaryDecoders.decodeReusingThreadLocalBinaryDecoder(data, schema, conversions);
22-
}
19+
static GenericRecord bytesToRecord(
20+
byte[] data, Schema schema, Collection<Conversion<?>> conversions) {
21+
return AvroBinaryDecoders.decodeReusingThreadLocalBinaryDecoder(data, schema, conversions);
22+
}
2323

24-
static byte[] recordToBytes(GenericRecord genericRecord, Schema schema) throws IOException {
25-
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
26-
BinaryEncoder binaryEncoder = EncoderFactory.get().binaryEncoder(outputStream, null);
27-
new GenericDatumWriter<>(schema).write(genericRecord, binaryEncoder);
28-
binaryEncoder.flush();
24+
static byte[] recordToBytes(GenericRecord genericRecord, Schema schema) throws IOException {
25+
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
26+
BinaryEncoder binaryEncoder = EncoderFactory.get().binaryEncoder(outputStream, null);
27+
new GenericDatumWriter<>(schema).write(genericRecord, binaryEncoder);
28+
binaryEncoder.flush();
2929

30-
return outputStream.toByteArray();
31-
}
30+
return outputStream.toByteArray();
31+
}
3232
}

‎hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/converter/AvroToJsonMessageConverter.java

+45-45
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,11 @@
11
package pl.allegro.tech.hermes.consumers.consumer.converter;
22

3+
import static java.util.stream.Collectors.toList;
4+
import static pl.allegro.tech.hermes.common.message.converter.AvroRecordToBytesConverter.bytesToRecord;
5+
import static pl.allegro.tech.hermes.common.message.wrapper.AvroMetadataMarker.METADATA_MARKER;
6+
import static pl.allegro.tech.hermes.consumers.consumer.Message.message;
7+
8+
import java.util.List;
39
import org.apache.avro.Schema;
410
import org.apache.avro.generic.GenericRecord;
511
import org.apache.avro.generic.GenericRecordBuilder;
@@ -9,51 +15,45 @@
915
import tech.allegro.schema.json2avro.converter.AvroJsonConverter;
1016
import tech.allegro.schema.json2avro.converter.conversions.DecimalAsStringConversion;
1117

12-
import java.util.List;
13-
14-
import static java.util.stream.Collectors.toList;
15-
import static pl.allegro.tech.hermes.common.message.converter.AvroRecordToBytesConverter.bytesToRecord;
16-
import static pl.allegro.tech.hermes.common.message.wrapper.AvroMetadataMarker.METADATA_MARKER;
17-
import static pl.allegro.tech.hermes.consumers.consumer.Message.message;
18-
1918
public class AvroToJsonMessageConverter implements MessageConverter {
2019

21-
private final AvroJsonConverter converter;
22-
23-
public AvroToJsonMessageConverter() {
24-
this.converter = new AvroJsonConverter(DecimalAsStringConversion.INSTANCE);
25-
}
26-
27-
@Override
28-
public Message convert(Message message, Topic topic) {
29-
return message()
30-
.fromMessage(message)
31-
.withContentType(ContentType.JSON)
32-
.withData(
33-
converter.convertToJson(
34-
recordWithoutMetadata(
35-
message.getData(), message.<Schema>getSchema().get().getSchema())))
36-
.withNoSchema()
37-
.build();
38-
}
39-
40-
private GenericRecord recordWithoutMetadata(byte[] data, Schema schema) {
41-
GenericRecord original = bytesToRecord(data, schema, List.of(DecimalAsStringConversion.INSTANCE));
42-
Schema schemaWithoutMetadata = removeMetadataField(schema);
43-
GenericRecordBuilder builder = new GenericRecordBuilder(schemaWithoutMetadata);
44-
schemaWithoutMetadata
45-
.getFields()
46-
.forEach(field -> builder.set(field, original.get(field.name())));
47-
return builder.build();
48-
}
49-
50-
private Schema removeMetadataField(Schema schema) {
51-
return Schema.createRecord(
52-
schema.getFields().stream()
53-
.filter(field -> !METADATA_MARKER.equals(field.name()))
54-
.map(
55-
field ->
56-
new Schema.Field(field.name(), field.schema(), field.doc(), field.defaultVal()))
57-
.collect(toList()));
58-
}
20+
private final AvroJsonConverter converter;
21+
22+
public AvroToJsonMessageConverter() {
23+
this.converter = new AvroJsonConverter(DecimalAsStringConversion.INSTANCE);
24+
}
25+
26+
@Override
27+
public Message convert(Message message, Topic topic) {
28+
return message()
29+
.fromMessage(message)
30+
.withContentType(ContentType.JSON)
31+
.withData(
32+
converter.convertToJson(
33+
recordWithoutMetadata(
34+
message.getData(), message.<Schema>getSchema().get().getSchema())))
35+
.withNoSchema()
36+
.build();
37+
}
38+
39+
private GenericRecord recordWithoutMetadata(byte[] data, Schema schema) {
40+
GenericRecord original =
41+
bytesToRecord(data, schema, List.of(DecimalAsStringConversion.INSTANCE));
42+
Schema schemaWithoutMetadata = removeMetadataField(schema);
43+
GenericRecordBuilder builder = new GenericRecordBuilder(schemaWithoutMetadata);
44+
schemaWithoutMetadata
45+
.getFields()
46+
.forEach(field -> builder.set(field, original.get(field.name())));
47+
return builder.build();
48+
}
49+
50+
private Schema removeMetadataField(Schema schema) {
51+
return Schema.createRecord(
52+
schema.getFields().stream()
53+
.filter(field -> !METADATA_MARKER.equals(field.name()))
54+
.map(
55+
field ->
56+
new Schema.Field(field.name(), field.schema(), field.doc(), field.defaultVal()))
57+
.collect(toList()));
58+
}
5959
}

‎hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/message/MessageToJsonConverter.java

+21-21
Original file line numberDiff line numberDiff line change
@@ -1,33 +1,33 @@
11
package pl.allegro.tech.hermes.frontend.publishing.message;
22

3+
import static pl.allegro.tech.hermes.common.message.converter.AvroRecordToBytesConverter.bytesToRecord;
4+
35
import org.apache.avro.Schema;
46
import pl.allegro.tech.hermes.common.message.wrapper.SchemaAwareSerDe;
57
import pl.allegro.tech.hermes.schema.CompiledSchema;
68
import tech.allegro.schema.json2avro.converter.AvroJsonConverter;
79

8-
import static pl.allegro.tech.hermes.common.message.converter.AvroRecordToBytesConverter.bytesToRecord;
9-
1010
public class MessageToJsonConverter {
11-
private final AvroJsonConverter converter = new AvroJsonConverter();
11+
private final AvroJsonConverter converter = new AvroJsonConverter();
1212

13-
public byte[] convert(Message message, boolean schemaIdAwareSerializationEnabled) {
14-
try {
15-
return message
16-
.<Schema>getCompiledSchema()
17-
.map(
18-
schema -> convertToJson(message.getData(), schema, schemaIdAwareSerializationEnabled))
19-
.orElseGet(message::getData);
20-
} catch (Exception ignored) {
21-
return message.getData();
22-
}
13+
public byte[] convert(Message message, boolean schemaIdAwareSerializationEnabled) {
14+
try {
15+
return message
16+
.<Schema>getCompiledSchema()
17+
.map(
18+
schema -> convertToJson(message.getData(), schema, schemaIdAwareSerializationEnabled))
19+
.orElseGet(message::getData);
20+
} catch (Exception ignored) {
21+
return message.getData();
2322
}
23+
}
2424

25-
private byte[] convertToJson(
26-
byte[] avro, CompiledSchema<Schema> schema, boolean schemaIdAwareSerializationEnabled) {
27-
byte[] schemaAwareAvro =
28-
schemaIdAwareSerializationEnabled
29-
? SchemaAwareSerDe.trimMagicByteAndSchemaVersion(avro)
30-
: avro;
31-
return converter.convertToJson(bytesToRecord(schemaAwareAvro, schema.getSchema()));
32-
}
25+
private byte[] convertToJson(
26+
byte[] avro, CompiledSchema<Schema> schema, boolean schemaIdAwareSerializationEnabled) {
27+
byte[] schemaAwareAvro =
28+
schemaIdAwareSerializationEnabled
29+
? SchemaAwareSerDe.trimMagicByteAndSchemaVersion(avro)
30+
: avro;
31+
return converter.convertToJson(bytesToRecord(schemaAwareAvro, schema.getSchema()));
32+
}
3333
}

‎hermes-management/src/main/java/pl/allegro/tech/hermes/management/config/AvroConfiguration.java

+8-8
Original file line numberDiff line numberDiff line change
@@ -8,13 +8,13 @@
88
@Configuration
99
public class AvroConfiguration {
1010

11-
@Bean
12-
AvroJsonConverter avroJsonConverter() {
13-
return new AvroJsonConverter();
14-
}
11+
@Bean
12+
AvroJsonConverter avroJsonConverter() {
13+
return new AvroJsonConverter();
14+
}
1515

16-
@Bean
17-
JsonAvroConverter jsonAvroConverter() {
18-
return new JsonAvroConverter();
19-
}
16+
@Bean
17+
JsonAvroConverter jsonAvroConverter() {
18+
return new JsonAvroConverter();
19+
}
2020
}

0 commit comments

Comments
 (0)