Our Community is getting an upgrade! To get everything ready for the relaunch, we’ll be placing the site in read-only mode starting September 21st.
We really appreciate your understanding while we get things set up behind the scenes. Catch up on all the exciting details about the move here.
Need help or have questions? Drop us a line at [email protected]
Created on 11-28-2022 12:24 PM - edited 11-28-2022 12:32 PM
Hi,
There is a kafka producer converting decimals to bytes using this:
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.nio.ByteBuffer;
import org.apache.avro.Conversion;
import org.apache.avro.Conversions;
import org.apache.avro.LogicalType;
import org.apache.avro.LogicalTypes;
import org.mapstruct.Mapper;
@Mapper
public class BigDecimalMapper {
public ByteBuffer bigDecimalToByteBuffer(BigDecimal bigDecimal) {
if (bigDecimal == null) return null;
Conversion<BigDecimal> conversion = new Conversions.DecimalConversion();
LogicalType type = LogicalTypes.decimal(15, 4);
BigDecimal scaledBigDecimal = bigDecimal.setScale(4, RoundingMode.HALF_UP);
// Schema parameter is not used by Conversions so pass as null
return conversion.toBytes(scaledBigDecimal, null, type);
}
We are using the confluent schema registry and this is the schema:
{
"type": "record",
"namespace": "event",
"name": "Event",
"fields": [
{
"name": "maxApprovedAmount",
"type": "bytes",
"logicalType": "decimal",
"precision": 4,
"scale": 4
}]
}
In Nifi I am using ConsumeKafkaRecord_2_6. using AvroReader and AvroRecordSetWriter.
In the reader I am using the schema from confluent. On the writer I am using Inherent Record Schema. When viewing the data in view flow file and selecting formatted, i see the value as:
"maxApprovedAmount" : {
"bytes" : "\u0001}x@"
},
Trouble is how to get the actual value and not this. I have tried multiple changes to my avro writer, and I have read other posts on similar issues but not able to get this resolved. I believe it has to with the lack of support for Logical types. I am even up to write a script in jython to do the conversion if it will work. Any help would be greatly appreciated!
Created 11-29-2022 09:01 AM
I am still unable to see how to achieve this. If anyone has any suggestions, I am open to it. Thanks!
Created 11-29-2022 01:36 PM
I found that the schema was defined incorrectly causing an issue. It should be
"name": "maxApprovedAmount",
"type": "bytes",
"logicalType": "decimal",
"precision": 4,
"scale": 4 "name": "maxApprovedAmount",
"type": {type: "bytes",
"logicalType": "decimal",
"precision": 4,
"scale": 4}