Skip to content

Commit affe0b2

Browse files
umayr-codesumairofficial
umayr-codes
and
umairofficial
authored
Source Codecs | Avro Codec follow-on PR (#2715)
* -Support for Source Codecs Signed-off-by: umairofficial <umairhusain1010@gmail.com> * -Support for Source Codecs Signed-off-by: umairofficial <umairhusain1010@gmail.com> * -Support for Sink Codecs Signed-off-by: umairofficial <umairhusain1010@gmail.com> * -Support for Sink Codecs Signed-off-by: umairofficial <umairhusain1010@gmail.com> --------- Co-authored-by: umairofficial <umairhusain1010@gmail.com>
1 parent c2d7767 commit affe0b2

3 files changed

Lines changed: 58 additions & 60 deletions

File tree

‎data-prepper-plugins/avro-codecs/build.gradle‎

Lines changed: 0 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,6 @@
22
* Copyright OpenSearch Contributors
33
* SPDX-License-Identifier: Apache-2.0
44
*/
5-
plugins {
6-
id 'java'
7-
}
8-
9-
repositories {
10-
mavenCentral()
11-
}
125

136
dependencies {
147
implementation project(path: ':data-prepper-api')

‎data-prepper-plugins/avro-codecs/src/main/java/org/opensearch/dataprepper/plugins/codec/avro/AvroInputCodec.java‎

Lines changed: 23 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@
99
import org.apache.avro.generic.GenericDatumReader;
1010
import org.apache.avro.generic.GenericEnumSymbol;
1111
import org.apache.avro.generic.GenericRecord;
12-
import org.apache.avro.generic.GenericFixed;
1312

1413
import org.apache.avro.file.DataFileStream;
1514
import org.apache.avro.util.Utf8;
@@ -60,14 +59,10 @@ private void parseAvroStream(final InputStream inputStream, final Consumer<Recor
6059

6160
while (stream.hasNext()) {
6261

63-
final Map<String, Object> eventData = new HashMap<>();
6462
GenericRecord avroRecord= stream.next();
6563

66-
for(Schema.Field field : schema.getFields()) {
67-
Object value=decodeValueIfEncoded(avroRecord.get(field.name()));
68-
eventData.put(field.name(), value);
64+
final Map<String, Object> eventData = convertRecordToMap(avroRecord, schema);
6965

70-
}
7166
final Event event = JacksonLog.builder().withData(eventData).build();
7267
eventConsumer.accept(new Record<>(event));
7368
}
@@ -78,20 +73,31 @@ private void parseAvroStream(final InputStream inputStream, final Consumer<Recor
7873
}
7974
}
8075

81-
private static Object decodeValueIfEncoded(Object rawValue){
82-
try{
83-
if(rawValue instanceof Utf8){
84-
byte[] utf8Bytes = rawValue.toString().getBytes("UTF-8");
85-
return new String(utf8Bytes, "UTF-8");
76+
private static Map<String, Object> convertRecordToMap(GenericRecord record, Schema schema) throws Exception {
77+
78+
final Map<String, Object> eventData = new HashMap<>();
79+
80+
for(Schema.Field field : schema.getFields()){
81+
82+
Object value = record.get(field.name());
83+
84+
if(value instanceof GenericRecord){
85+
Schema schemaOfNestedRecord = ((GenericRecord) value).getSchema();
86+
value = convertRecordToMap((GenericRecord) value, schemaOfNestedRecord);
8687
}
87-
else if(rawValue instanceof GenericEnumSymbol || rawValue instanceof GenericData.EnumSymbol || rawValue instanceof GenericFixed || rawValue instanceof GenericRecord){
88-
throw new Exception("The Avro codec does not support this data type presently");
88+
89+
else if(value instanceof GenericEnumSymbol || value instanceof GenericData.EnumSymbol){
90+
value = value.toString();
8991
}
90-
return rawValue;
91-
}
92-
catch (Exception e){
93-
return rawValue;
92+
93+
else if(value instanceof Utf8){
94+
byte[] utf8Bytes = value.toString().getBytes("UTF-8");
95+
value = new String(utf8Bytes, "UTF-8");
96+
}
97+
98+
eventData.put(field.name(), value);
9499
}
100+
return eventData;
95101
}
96102

97103
}

‎data-prepper-plugins/avro-codecs/src/test/java/org/opensearch/dataprepper/plugins/codec/avro/AvroInputCodecTest.java‎

Lines changed: 35 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -22,16 +22,13 @@
2222
import org.mockito.junit.jupiter.MockitoExtension;
2323
import org.opensearch.dataprepper.model.event.Event;
2424
import org.opensearch.dataprepper.model.event.EventType;
25-
import org.opensearch.dataprepper.model.log.JacksonLog;
2625
import org.opensearch.dataprepper.model.record.Record;
27-
import org.json.JSONObject;
2826

2927
import java.io.File;
3028
import java.io.IOException;
3129
import java.io.InputStream;
3230
import java.io.FileInputStream;
3331
import java.io.ByteArrayInputStream;
34-
import java.nio.charset.StandardCharsets;
3532
import java.nio.file.Files;
3633
import java.nio.file.Path;
3734
import java.nio.file.Paths;
@@ -139,45 +136,28 @@ public void test_HappyCaseAvroInputStream_then_callsConsumerWithParsedEvents(fin
139136
assertThat(actualRecord.getData().getMetadata(),notNullValue());
140137
assertThat(actualRecord.getData().getMetadata().getEventType(), equalTo(EventType.LOG.toString()));
141138

142-
Map expectedMap=getEvent(index).toMap();
143-
assertThat(actualRecord.getData().toMap(), equalTo(expectedMap));
139+
Map<String,Object> expectedMap=new HashMap<>();
140+
GenericRecord record=generateRecords(parseSchema(),numberOfRecords).get(index);
141+
for(Schema.Field field:record.getSchema().getFields()){
142+
expectedMap.put(field.name(),record.get(field.name()));
143+
}
144144

145-
for(Object key: actualRecord.getData().toMap().keySet()){
146-
Object decodedOutput = decodeOutputIfEncoded(actualRecord.getData().toMap() , key);
147-
Object expectedOutput = getEvent(index).toMap().get(key.toString());
148-
assertThat(decodedOutput, equalTo(expectedOutput));
145+
for(String key: expectedMap.keySet()){
146+
Object actualRecordValue=actualRecord.getData().toMap().get(key);
147+
if(!(actualRecordValue instanceof Map))
148+
assertThat(actualRecord.getData().toMap().get(key), equalTo(expectedMap.get(key)));
149+
else{
150+
GenericRecord expectedInnerRecord= (GenericRecord) expectedMap.get(key);
151+
Schema innerSchema=expectedInnerRecord.getSchema();
152+
for(Schema.Field innerField : innerSchema.getFields()){
153+
assertThat(((Map)actualRecordValue).get(innerField.name()),equalTo(expectedInnerRecord.get(innerField.name())));
154+
}
155+
}
149156
}
150157
index++;
151158
}
152159
fileInputStream.close();
153160
Files.delete(path);
154-
155-
}
156-
157-
private static Object decodeOutputIfEncoded(Map encodedOutput, Object key){
158-
try{
159-
JSONObject outputJson = new JSONObject(encodedOutput);
160-
Map innerJson= (Map) outputJson.get(key.toString());
161-
byte[] encodedString=(byte[]) innerJson.get("bytes");
162-
return new String(encodedString, StandardCharsets.UTF_8);
163-
164-
}catch (Exception e){
165-
return encodedOutput.get(key);
166-
}
167-
}
168-
169-
private static Event getEvent(int index){
170-
List<GenericRecord> recordList=generateRecords(parseSchema(),numberOfRecords);
171-
GenericRecord record=recordList.get(index);
172-
Schema schema=parseSchema();
173-
final Map<String, Object> eventData = new HashMap<>();
174-
for(Schema.Field field : schema.getFields()) {
175-
176-
eventData.put(field.name(), record.get(field.name()));
177-
178-
}
179-
final Event event = JacksonLog.builder().withData(eventData).build();
180-
return event;
181161
}
182162

183163

@@ -211,9 +191,13 @@ private static List<GenericRecord> generateRecords(Schema schema, int numberOfRe
211191
for(int rows = 0; rows < numberOfRecords; rows++){
212192

213193
GenericRecord record = new GenericData.Record(schema);
194+
GenericRecord innerRecord = new GenericData.Record(parseInnerSchemaForNestedRecord());
195+
innerRecord.put("firstFieldInNestedRecord", "testString"+rows);
196+
innerRecord.put("secondFieldInNestedRecord", rows);
214197

215198
record.put("name", "Person"+rows);
216199
record.put("age", rows);
200+
record.put("nestedRecord", innerRecord);
217201
recordList.add((record));
218202

219203
}
@@ -224,14 +208,29 @@ private static List<GenericRecord> generateRecords(Schema schema, int numberOfRe
224208

225209
private static Schema parseSchema() {
226210

211+
Schema innerSchema=parseInnerSchemaForNestedRecord();
227212
return SchemaBuilder.record("Person")
228213
.fields()
229214
.name("name").type().stringType().noDefault()
230215
.name("age").type().intType().noDefault()
216+
.name("nestedRecord").type(innerSchema).noDefault()
231217
.endRecord();
232218

233219
}
234220

221+
private static Schema parseInnerSchemaForNestedRecord(){
222+
return SchemaBuilder
223+
.record("InnerRecord")
224+
.fields()
225+
.name("firstFieldInNestedRecord")
226+
.type(Schema.create(Schema.Type.STRING))
227+
.noDefault()
228+
.name("secondFieldInNestedRecord")
229+
.type(Schema.create(Schema.Type.INT))
230+
.noDefault()
231+
.endRecord();
232+
}
233+
235234
private static InputStream createInvalidAvroStream() {
236235
return new ByteArrayInputStream(INVALID_AVRO_INPUT_STREAM.getBytes());
237236
}

0 commit comments

Comments
 (0)