@@ -0,0 +1,3302 @@
+/**%0A * Licensed to the Apache Software Foundation (ASF) under one%0A * or more contributor license agreements.  See the NOTICE file%0A * distributed with this work for additional information%0A * regarding copyright ownership.  The ASF licenses this file%0A * to you under the Apache License, Version 2.0 (the%0A * %22License%22); you may not use this file except in compliance%0A * with the License.  You may obtain a copy of the License at%0A *%0A *   http://www.apache.org/licenses/LICENSE-2.0%0A *%0A * Unless required by applicable law or agreed to in writing,%0A * software distributed under the License is distributed on an%0A * %22AS IS%22 BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY%0A * KIND, either express or implied.  See the License for the%0A * specific language governing permissions and limitations%0A * under the License.%0A */%0Apackage org.apache.pinot.core.realtime.impl.kafka;%0A%0Aimport com.google.common.base.Preconditions;%0Aimport java.io.IOException;%0Aimport java.util.Map;%0Aimport javax.annotation.concurrent.NotThreadSafe;%0Aimport org.apache.avro.generic.GenericData;%0Aimport org.apache.avro.generic.GenericDatumReader;%0Aimport org.apache.avro.io.BinaryDecoder;%0Aimport org.apache.avro.io.DatumReader;%0Aimport org.apache.avro.io.DecoderFactory;%0Aimport org.apache.pinot.common.data.Schema;%0Aimport org.apache.pinot.core.data.GenericRow;%0Aimport org.apache.pinot.core.realtime.stream.StreamMessageDecoder;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0A%0A@NotThreadSafe%0Apublic class SimpleAvroMessageDecoder implements StreamMessageDecoder%3Cbyte%5B%5D%3E %7B%0A  private static final Logger LOGGER = LoggerFactory.getLogger(SimpleAvroMessageDecoder.class);%0A%0A  private static final String SCHEMA = %22schema%22;%0A%0A  private org.apache.avro.Schema _avroSchema;%0A  private DatumReader%3CGenericData.Record%3E _datumReader;%0A  private AvroRecordToPinotRowGenerator _avroRecordConverter;%0A  private BinaryDecoder _binaryDecoderToReuse;%0A  private GenericData.Record _avroRecordToReuse;%0A%0A  @Override%0A  public void init(Map%3CString, String%3E props, Schema indexingSchema, String topicName)%0A      throws Exception %7B%0A    Preconditions.checkState(props.containsKey(SCHEMA), %22Avro schema must be provided%22);%0A    _avroSchema = new org.apache.avro.Schema.Parser().parse(props.get(SCHEMA));%0A    _datumReader = new GenericDatumReader%3C%3E(_avroSchema);%0A    _avroRecordConverter = new AvroRecordToPinotRowGenerator(indexingSchema);%0A  %7D%0A%0A  /**%0A   * %7B@inheritDoc%7D%0A   *%0A   * %3Cp%3ENOTE: the payload should contain message content only (without header).%0A   */%0A  @Override%0A  public GenericRow decode(byte%5B%5D payload, GenericRow destination) %7B%0A    return decode(payload, 0, payload.length, destination);%0A  %7D%0A%0A  /**%0A   * %7B@inheritDoc%7D%0A   *%0A   * %3Cp%3ENOTE: the payload should contain message content only (without header).%0A   */%0A  @Override%0A  public GenericRow decode(byte%5B%5D payload, int offset, int length, GenericRow destination) %7B%0A    _binaryDecoderToReuse = DecoderFactory.get().binaryDecoder(payload, offset, length, _binaryDecoderToReuse);%0A    try %7B%0A      _avroRecordToReuse = _datumReader.read(_avroRecordToReuse, _binaryDecoderToReuse);%0A    %7D catch (IOException e) %7B%0A      LOGGER.error(%22Caught exception while reading message using schema: %7B%7D%22, _avroSchema, e);%0A      return null;%0A    %7D%0A    return _avroRecordConverter.transform(_avroRecordToReuse, destination);%0A  %7D%0A%7D%0A

Add SimpleAvroMessageDecoder which allows passing in Avro schema directly (#4197)