DRILL-8521: Upgrade parquet-avro to 1.15.1, Avro to 1.12.0 (#2986)

* Upgrade parquet-avro to 1.15.1
* Upgrade Avro to 1.12.0
* Upgrade kafka-avro-serializer to 7.9.0

Author: rymarm

contrib/storage-hive/hive-exec-shade/pom.xml

@@ -32,7 +32,7 @@
   <name>Drill : Contrib : Storage : Hive : Exec Shaded</name>
 
   <properties>
-    <hive.parquet.version>1.8.3</hive.parquet.version>
+    <hive.parquet.version>1.15.1</hive.parquet.version>
   </properties>
 
   <dependencyManagement>
@@ -219,6 +219,12 @@
                 <exclude>META-INF/versions/22/**</exclude>
               </excludes>
             </filter>
+	    <filter>
+              <artifact>org.apache.parquet:parquet-hadoop-bundle</artifact>
+              <excludes>
+                <exclude>META-INF/versions/22/**</exclude>
+              </excludes>
+            </filter>
           </filters>
         </configuration>
       </plugin>

contrib/storage-kafka/pom.xml

@@ -97,7 +97,7 @@
     <dependency>
       <groupId>io.confluent</groupId>
       <artifactId>kafka-avro-serializer</artifactId>
-      <version>6.1.1</version>
+      <version>7.9.0</version>
     </dependency>
 
     <!-- Test dependencies -->

exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ColumnDataReader.java

@@ -69,7 +69,7 @@
 import org.apache.parquet.column.values.factory.DefaultV2ValuesWriterFactory;
 import org.apache.parquet.column.values.factory.ValuesWriterFactory;
 import org.apache.parquet.compression.CompressionCodecFactory;
-import org.apache.parquet.hadoop.ParquetColumnChunkPageWriteStore;
+import org.apache.parquet.hadoop.ColumnChunkPageWriteStore;
 import org.apache.parquet.hadoop.ParquetFileWriter;
 import org.apache.parquet.hadoop.metadata.CompressionCodecName;
 import org.apache.parquet.io.ColumnIOFactory;
@@ -120,7 +120,7 @@ public class ParquetRecordWriter extends ParquetOutputRecordWriter {
   private long recordCountForNextMemCheck = MINIMUM_RECORD_COUNT_FOR_CHECK;
 
   private ColumnWriteStore store;
-  private ParquetColumnChunkPageWriteStore pageStore;
+  private ColumnChunkPageWriteStore pageStore;
 
   private RecordConsumer consumer;
   private BatchSchema batchSchema;
@@ -285,10 +285,7 @@ private void newSchema() {
         .withWriterVersion(writerVersion)
         .build();
 
-    // TODO: Replace ParquetColumnChunkPageWriteStore with ColumnChunkPageWriteStore from parquet library
-    //   once DRILL-7906 (PARQUET-1006) will be resolved
-    pageStore = new ParquetColumnChunkPageWriteStore(codecFactory.getCompressor(codec), schema,
-            parquetProperties.getInitialSlabSize(), pageSize, parquetProperties.getAllocator(),
+    pageStore = new ColumnChunkPageWriteStore(codecFactory.getCompressor(codec), schema, parquetProperties.getAllocator(),
             parquetProperties.getColumnIndexTruncateLength(), parquetProperties.getPageWriteChecksumEnabled());
 
     store = writerVersion == WriterVersion.PARQUET_1_0

exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRecordWriter.java

@@ -55,7 +55,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -417,13 +416,9 @@ public void close() {
     recordReader = null;
     recordMaterializer = null;
     nullFilledVectors = null;
-    try {
-      if (pageReadStore != null) {
-        pageReadStore.close();
-        pageReadStore = null;
-      }
-    } catch (IOException e) {
-      logger.warn("Failure while closing PageReadStore", e);
+    if (pageReadStore != null) {
+      pageReadStore.close();
+      pageReadStore = null;
     }
   }
 

exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet2/DrillParquetReader.java

@@ -295,9 +295,13 @@ public void addColumn(ColumnDescriptor descriptor, ColumnChunkMetaData metaData)
     columns.put(descriptor, reader);
   }
 
-  public void close() throws IOException {
+  public void close() {
     for (FSDataInputStream stream : streams) {
-      stream.close();
+      try {
+        stream.close();
+      } catch (IOException e) {
+        logger.warn("Error closing stream: {}", e.getMessage(), e);
+      }
     }
     for (ColumnChunkIncPageReader reader : columns.values()) {
       reader.close();