Merge branch '5.0.4-post' into 5.1.0-post

albertochong · Jun 4, 2020 · 2a72d6e · 2a72d6e
2 parents 289c01e + be647a0
commit 2a72d6e
Show file tree

Hide file tree

Showing 5 changed files with 15 additions and 9 deletions.
diff --git a/src/main/java/io/confluent/examples/streams/microservices/EmailService.java b/src/main/java/io/confluent/examples/streams/microservices/EmailService.java
@@ -135,8 +135,7 @@ interface Emailer {
     void sendEmail(EmailTuple details);
   }
 
-  public class EmailTuple {
-
+  public static class EmailTuple {
     public Order order;
     public Payment payment;
     public Customer customer;

diff --git a/src/main/java/io/confluent/examples/streams/microservices/OrdersService.java b/src/main/java/io/confluent/examples/streams/microservices/OrdersService.java
@@ -181,7 +181,7 @@ public void getWithTimeout(@PathParam("id") final String id,
     }
   }
 
-  class FilteredResponse<K, V> {
+  static class FilteredResponse<K, V> {
     private final AsyncResponse asyncResponse;
     private final Predicate<K, V> predicate;
 
@@ -236,7 +236,7 @@ private HostStoreInfo getKeyLocationOrBlock(final String id, final AsyncResponse
       }
       try {
         //Sleep a bit until metadata becomes available
-        Thread.sleep(Math.min(Long.valueOf(CALL_TIMEOUT), 200));
+        Thread.sleep(Math.min(Long.parseLong(CALL_TIMEOUT), 200));
       } catch (final InterruptedException e) {
         e.printStackTrace();
       }
@@ -264,6 +264,7 @@ private void fetchFromOtherHost(final String path, final AsyncResponse asyncResp
           });
       asyncResponse.resume(bean);
     } catch (final Exception swallowed) {
+      log.warn("GET failed.", swallowed);
     }
   }
 
@@ -410,7 +411,7 @@ public static void main(final String[] args) throws Exception {
     final String restPort = args.length > 3 ? args[3] : null;
 
     Schemas.configureSerdesWithSchemaRegistryUrl(schemaRegistryUrl);
-    final OrdersService service = new OrdersService(restHostname, restPort == null ? 0 : Integer.valueOf(restPort));
+    final OrdersService service = new OrdersService(restHostname, restPort == null ? 0 : Integer.parseInt(restPort));
     service.start(bootstrapServers, "/tmp/kafka-streams");
     addShutdownHookAndBlock(service);
   }

diff --git a/src/main/java/io/confluent/examples/streams/microservices/domain/Schemas.java b/src/main/java/io/confluent/examples/streams/microservices/domain/Schemas.java
@@ -59,7 +59,7 @@ public String toString() {
 
   public static class Topics {
 
-    public static final Map<String, Topic<?, ?>> ALL = new HashMap<>();
+    public final static Map<String, Topic<?, ?>> ALL = new HashMap<>();
     public static Topic<String, Order> ORDERS;
     public static Topic<String, OrderEnriched> ORDERS_ENRICHED;
     public static Topic<String, Payment> PAYMENTS;

diff --git a/src/main/java/io/confluent/examples/streams/microservices/util/MicroserviceUtils.java b/src/main/java/io/confluent/examples/streams/microservices/util/MicroserviceUtils.java
@@ -4,6 +4,8 @@
 import io.confluent.examples.streams.microservices.Service;
 import io.confluent.examples.streams.microservices.domain.Schemas;
 import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
+
+import org.apache.commons.compress.utils.Charsets;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -24,6 +26,7 @@
 
 import javax.ws.rs.container.AsyncResponse;
 import javax.ws.rs.core.Response;
+import java.nio.charset.Charset;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
@@ -115,7 +118,7 @@ public void configure(final Map<String, ?> map, final boolean b) {
 
         @Override
         public byte[] serialize(final String topic, final Product pt) {
-          return pt.toString().getBytes();
+          return pt.toString().getBytes(Charsets.UTF_8);
         }
 
         @Override
@@ -133,7 +136,7 @@ public void configure(final Map<String, ?> map, final boolean b) {
 
         @Override
         public Product deserialize(final String topic, final byte[] bytes) {
-          return Product.valueOf(new String(bytes));
+          return Product.valueOf(new String(bytes, Charsets.UTF_8));
         }
 
         @Override

diff --git a/src/main/java/io/confluent/examples/streams/utils/PriorityQueueDeserializer.java b/src/main/java/io/confluent/examples/streams/utils/PriorityQueueDeserializer.java
@@ -20,6 +20,7 @@
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
 import java.io.IOException;
+import java.nio.BufferUnderflowException;
 import java.util.Comparator;
 import java.util.Map;
 import java.util.PriorityQueue;
@@ -50,7 +51,9 @@ public PriorityQueue<T> deserialize(final String s, final byte[] bytes) {
             final int records = dataInputStream.readInt();
             for (int i = 0; i < records; i++) {
                 final byte[] valueBytes = new byte[dataInputStream.readInt()];
-                dataInputStream.read(valueBytes);
+                if (dataInputStream.read(valueBytes) != valueBytes.length) {
+                    throw new BufferUnderflowException();
+                };
                 priorityQueue.add(valueDeserializer.deserialize(s, valueBytes));
             }
         } catch (final IOException e) {