Fix exception handling

confluentinc · Aug 10, 2023 · 9ef14fc · 9ef14fc
1 parent 049ba7b
commit 9ef14fc
Show file tree

Hide file tree

Showing 27 changed files with 76 additions and 24 deletions.
diff --git a/avro-serializer/src/main/java/io/confluent/kafka/formatter/AvroMessageFormatter.java b/avro-serializer/src/main/java/io/confluent/kafka/formatter/AvroMessageFormatter.java
@@ -134,7 +134,7 @@ public Object deserialize(String topic, byte[] payload) throws SerializationExce
     }
 
     @Override
-    public void close() {
+    public void close() throws IOException {
       if (keyDeserializer != null) {
         keyDeserializer.close();
       }

diff --git a/avro-serializer/src/main/java/io/confluent/kafka/formatter/AvroMessageReader.java b/avro-serializer/src/main/java/io/confluent/kafka/formatter/AvroMessageReader.java
@@ -173,7 +173,7 @@ public byte[] serialize(
     }
 
     @Override
-    public void close() {
+    public void close() throws IOException {
       if (keySerializer != null) {
         keySerializer.close();
       }

diff --git a/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroDeserializer.java b/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroDeserializer.java
@@ -19,6 +19,7 @@
 import org.apache.avro.Schema;
 import org.apache.kafka.common.serialization.Deserializer;
 
+import java.io.IOException;
 import java.util.Map;
 
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
@@ -62,6 +63,10 @@ public Object deserialize(String topic, byte[] bytes, Schema readerSchema) {
 
   @Override
   public void close() {
-    super.close();
+    try {
+      super.close();
+    } catch (IOException e) {
+      throw new RuntimeException("Exception while closing deserializer", e);
+    }
   }
 }
diff --git a/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroSerializer.java b/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroSerializer.java
@@ -18,6 +18,7 @@
 
 import org.apache.kafka.common.serialization.Serializer;
 
+import java.io.IOException;
 import java.util.Map;
 
 import io.confluent.kafka.schemaregistry.avro.AvroSchema;
@@ -64,6 +65,10 @@ public byte[] serialize(String topic, Object record) {
 
   @Override
   public void close() {
-    super.close();
+    try {
+      super.close();
+    } catch (IOException e) {
+      throw new RuntimeException("Exception while closing serializer", e);
+    }
   }
 }
diff --git a/...nt/src/main/java/io/confluent/kafka/schemaregistry/client/CachedSchemaRegistryClient.java b/...nt/src/main/java/io/confluent/kafka/schemaregistry/client/CachedSchemaRegistryClient.java
@@ -683,7 +683,7 @@ public synchronized void reset() {
   }
 
   @Override
-  public void close() {
+  public void close() throws IOException {
     if (restService != null) {
       restService.close();
     }

diff --git a/client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/RestService.java b/client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/RestService.java
@@ -1255,7 +1255,7 @@ public void setProxy(String proxyHost, int proxyPort) {
   }
 
   @Override
-  public void close() {
+  public void close() throws IOException {
     if (bearerAuthCredentialProvider != null) {
       bearerAuthCredentialProvider.close();
     }

diff --git a/...nfluent/kafka/schemaregistry/client/security/bearerauth/BearerAuthCredentialProvider.java b/...nfluent/kafka/schemaregistry/client/security/bearerauth/BearerAuthCredentialProvider.java
@@ -19,6 +19,7 @@
 import org.apache.kafka.common.Configurable;
 
 import java.io.Closeable;
+import java.io.IOException;
 import java.net.URL;
 
 public interface BearerAuthCredentialProvider extends Closeable, Configurable {

diff --git a/core/src/main/java/io/confluent/kafka/schemaregistry/rest/SchemaRegistryRestApplication.java b/core/src/main/java/io/confluent/kafka/schemaregistry/rest/SchemaRegistryRestApplication.java
@@ -141,7 +141,11 @@ protected ResourceCollection getStaticResources() {
   public void onShutdown() {
 
     if (schemaRegistry != null) {
-      schemaRegistry.close();
+      try {
+        schemaRegistry.close();
+      } catch (IOException e) {
+        log.error("Error closing schema registry", e);
+      }
     }
 
     if (schemaRegistryResourceExtensions != null) {

diff --git a/core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaSchemaRegistry.java b/core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaSchemaRegistry.java
@@ -1516,7 +1516,7 @@ private CloseableIterator<SchemaRegistryValue> allVersionsFromAllContexts(
   }
 
   @Override
-  public void close() {
+  public void close() throws IOException {
     log.info("Shutting down schema registry");
     kafkaStore.close();
     if (leaderElector != null) {

diff --git a/core/src/main/java/io/confluent/kafka/schemaregistry/storage/SchemaRegistry.java b/core/src/main/java/io/confluent/kafka/schemaregistry/storage/SchemaRegistry.java
@@ -15,6 +15,7 @@
 
 package io.confluent.kafka.schemaregistry.storage;
 
+import java.io.IOException;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -91,7 +92,7 @@ List<String> isCompatible(String subject,
                             Schema newSchema,
                             List<Schema> previousSchemas) throws SchemaRegistryException;
 
-  void close();
+  void close() throws IOException;
 
   void deleteSchemaVersion(String subject, Schema schema,
                            boolean permanentDelete) throws SchemaRegistryException;

diff --git a/...rc/main/java/io/confluent/kafka/schemaregistry/tools/SchemaRegistryClientPerformance.java b/...rc/main/java/io/confluent/kafka/schemaregistry/tools/SchemaRegistryClientPerformance.java
@@ -85,7 +85,7 @@ protected void doIteration(PerformanceStats.Callback cb) {
   }
 
   @Override
-  protected void close() {
+  protected void close() throws IOException {
     if (client != null) {
       client.close();
     }

diff --git a/core/src/main/java/io/confluent/kafka/schemaregistry/tools/SchemaRegistryPerformance.java b/core/src/main/java/io/confluent/kafka/schemaregistry/tools/SchemaRegistryPerformance.java
@@ -143,7 +143,7 @@ protected void doIteration(PerformanceStats.Callback cb) {
     cb.onCompletion(1, 0);
   }
 
-  protected void close() {
+  protected void close() throws IOException {
     // We can see some failures due to things like timeouts, but we want it to be obvious
     // if there are too many failures (indicating a real underlying problem). 1% is an arbitrarily
     // chosen limit.

diff --git a/...erializer/src/main/java/io/confluent/kafka/formatter/json/JsonSchemaMessageFormatter.java b/...erializer/src/main/java/io/confluent/kafka/formatter/json/JsonSchemaMessageFormatter.java
@@ -131,7 +131,7 @@ public JsonNode deserialize(String topic, byte[] payload) throws SerializationEx
     }
 
     @Override
-    public void close() {
+    public void close() throws IOException {
       if (keyDeserializer != null) {
         keyDeserializer.close();
       }

diff --git a/...a-serializer/src/main/java/io/confluent/kafka/formatter/json/JsonSchemaMessageReader.java b/...a-serializer/src/main/java/io/confluent/kafka/formatter/json/JsonSchemaMessageReader.java
@@ -170,7 +170,7 @@ public byte[] serialize(
     }
 
     @Override
-    public void close() {
+    public void close() throws IOException {
       if (keySerializer != null) {
         keySerializer.close();
       }

diff --git a/...alizer/src/main/java/io/confluent/kafka/serializers/json/KafkaJsonSchemaDeserializer.java b/...alizer/src/main/java/io/confluent/kafka/serializers/json/KafkaJsonSchemaDeserializer.java
@@ -19,6 +19,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.kafka.common.serialization.Deserializer;
 
+import java.io.IOException;
 import java.util.Map;
 
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
@@ -81,6 +82,10 @@ public T deserialize(String topic, byte[] bytes) {
 
   @Override
   public void close() {
-    super.close();
+    try {
+      super.close();
+    } catch (IOException e) {
+      throw new RuntimeException("Exception while closing deserializer", e);
+    }
   }
 }
diff --git a/...rializer/src/main/java/io/confluent/kafka/serializers/json/KafkaJsonSchemaSerializer.java b/...rializer/src/main/java/io/confluent/kafka/serializers/json/KafkaJsonSchemaSerializer.java
@@ -90,6 +90,10 @@ private JsonSchema getSchema(T record) {
 
   @Override
   public void close() {
-    super.close();
+    try {
+      super.close();
+    } catch (IOException e) {
+      throw new RuntimeException("Exception while closing serializer", e);
+    }
   }
 }
diff --git a/...gin/src/main/java/io/confluent/kafka/schemaregistry/maven/DownloadSchemaRegistryMojo.java b/...gin/src/main/java/io/confluent/kafka/schemaregistry/maven/DownloadSchemaRegistryMojo.java
@@ -29,6 +29,7 @@
 
 import java.io.File;
 import java.io.FileOutputStream;
+import java.io.IOException;
 import java.io.OutputStreamWriter;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
@@ -175,7 +176,11 @@ public void execute() throws MojoExecutionException, MojoFailureException {
         );
       }
     }
-    close();
+    try {
+      close();
+    } catch (IOException e) {
+      throw new MojoExecutionException("Exception while closing schema registry client", e);
+    }
   }
 
   private String getExtension(ParsedSchema parsedSchema) {

diff --git a/maven-plugin/src/main/java/io/confluent/kafka/schemaregistry/maven/SchemaRegistryMojo.java b/maven-plugin/src/main/java/io/confluent/kafka/schemaregistry/maven/SchemaRegistryMojo.java
@@ -21,6 +21,7 @@
 import org.apache.maven.plugins.annotations.Parameter;
 
 import java.io.Closeable;
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -101,7 +102,7 @@ private List<SchemaProvider> defaultSchemaProviders() {
   }
 
   @Override
-  public void close() {
+  public void close() throws IOException {
     if (client != null) {
       client.close();
     }

diff --git a/...ializer/src/main/java/io/confluent/kafka/formatter/protobuf/ProtobufMessageFormatter.java b/...ializer/src/main/java/io/confluent/kafka/formatter/protobuf/ProtobufMessageFormatter.java
@@ -149,7 +149,7 @@ public Message deserialize(String topic, byte[] payload) throws SerializationExc
     }
 
     @Override
-    public void close() {
+    public void close() throws IOException {
       if (keyDeserializer != null) {
         keyDeserializer.close();
       }

diff --git a/...serializer/src/main/java/io/confluent/kafka/formatter/protobuf/ProtobufMessageReader.java b/...serializer/src/main/java/io/confluent/kafka/formatter/protobuf/ProtobufMessageReader.java
@@ -17,6 +17,8 @@
 
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Message;
+
+import java.io.IOException;
 import java.util.Properties;
 import org.apache.kafka.common.errors.SerializationException;
 import org.apache.kafka.common.serialization.Serializer;
@@ -158,7 +160,7 @@ public byte[] serialize(
     }
 
     @Override
-    public void close() {
+    public void close() throws IOException {
       if (keySerializer != null) {
         keySerializer.close();
       }

diff --git a/...izer/src/main/java/io/confluent/kafka/serializers/protobuf/KafkaProtobufDeserializer.java b/...izer/src/main/java/io/confluent/kafka/serializers/protobuf/KafkaProtobufDeserializer.java
@@ -20,6 +20,7 @@
 import com.google.protobuf.Message;
 import org.apache.kafka.common.serialization.Deserializer;
 
+import java.io.IOException;
 import java.util.Map;
 
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
@@ -78,6 +79,10 @@ public T deserialize(String topic, byte[] bytes) {
 
   @Override
   public void close() {
-    super.close();
+    try {
+      super.close();
+    } catch (IOException e) {
+      throw new RuntimeException("Exception while closing deserializer", e);
+    }
   }
 }
diff --git a/...alizer/src/main/java/io/confluent/kafka/serializers/protobuf/KafkaProtobufSerializer.java b/...alizer/src/main/java/io/confluent/kafka/serializers/protobuf/KafkaProtobufSerializer.java
@@ -103,6 +103,10 @@ public byte[] serialize(String topic, T record) {
 
   @Override
   public void close() {
-    super.close();
+    try {
+      super.close();
+    } catch (IOException e) {
+      throw new RuntimeException("Exception while closing serializer", e);
+    }
   }
 }
diff --git a/schema-serializer/src/main/java/io/confluent/kafka/formatter/SchemaMessageDeserializer.java b/schema-serializer/src/main/java/io/confluent/kafka/formatter/SchemaMessageDeserializer.java
@@ -20,6 +20,7 @@
 import org.apache.kafka.common.serialization.Deserializer;
 
 import java.io.Closeable;
+import java.io.IOException;
 
 public interface SchemaMessageDeserializer<T> extends Closeable {
 

diff --git a/schema-serializer/src/main/java/io/confluent/kafka/formatter/SchemaMessageFormatter.java b/schema-serializer/src/main/java/io/confluent/kafka/formatter/SchemaMessageFormatter.java
@@ -199,7 +199,11 @@ public void writeTo(ConsumerRecord<byte[], byte[]> consumerRecord, PrintStream o
   @Override
   public void close() {
     if (deserializer != null) {
-      deserializer.close();
+      try {
+        deserializer.close();
+      } catch (IOException e) {
+        throw new RuntimeException("Exception while closing deserializer", e);
+      }
     }
   }
 

diff --git a/schema-serializer/src/main/java/io/confluent/kafka/formatter/SchemaMessageReader.java b/schema-serializer/src/main/java/io/confluent/kafka/formatter/SchemaMessageReader.java
@@ -383,7 +383,11 @@ private byte[] serializeNonSchemaKey(String keyString) {
   @Override
   public void close() {
     if (serializer != null) {
-      serializer.close();
+      try {
+        serializer.close();
+      } catch (IOException e) {
+        throw new RuntimeException("Exception while closing serializer", e);
+      }
     }
   }
 }
diff --git a/schema-serializer/src/main/java/io/confluent/kafka/formatter/SchemaMessageSerializer.java b/schema-serializer/src/main/java/io/confluent/kafka/formatter/SchemaMessageSerializer.java
@@ -21,6 +21,7 @@
 import io.confluent.kafka.schemaregistry.ParsedSchema;
 
 import java.io.Closeable;
+import java.io.IOException;
 
 public interface SchemaMessageSerializer<T> extends Closeable {
 

diff --git a/schema-serializer/src/main/java/io/confluent/kafka/serializers/AbstractKafkaSchemaSerDe.java b/schema-serializer/src/main/java/io/confluent/kafka/serializers/AbstractKafkaSchemaSerDe.java
@@ -265,7 +265,7 @@ protected static KafkaException toKafkaException(RestClientException e, String e
   }
 
   @Override
-  public void close() {
+  public void close() throws IOException {
     if (schemaRegistry != null) {
       schemaRegistry.close();
     }