#vulnerable code 
                              @Override
  public ServiceMessage decodeMessage(Payload payload) {
    Builder builder = ServiceMessage.builder();
    
    if (payload.getData().hasRemaining()) {
      try {
        builder.data(payload.sliceData());
      } catch (Throwable ex) {
        LOGGER.error("Failed to deserialize data", ex);
      }
    }

    if (payload.hasMetadata()) {
      ByteBuf headers = payload.sliceMetadata();
      ByteBufInputStream inputStream = new ByteBufInputStream(headers);
      try {
        builder.headers((Map<String, String>) (readFrom(inputStream, mapType)));
      } catch (Throwable ex) {
        LOGGER.error("Failed to deserialize data", ex);
      }
    }


    return builder.build();
  } 
                            #location 17 
                            #vulnerability type RESOURCE_LEAK
#fixed code
                              @Override
  public ServiceMessage decodeMessage(Payload payload) {
    Builder builder = ServiceMessage.builder();

    if (payload.getData().hasRemaining()) {
      try {
        builder.data(payload.sliceData());
      } catch (Throwable ex) {
        LOGGER.error("Failed to deserialize data", ex);
      }
    }

    if (payload.hasMetadata()) {
      try (ByteBufInputStream inputStream = new ByteBufInputStream(payload.sliceMetadata(), true)) {
        builder.headers(readFrom(inputStream, mapType));
      } catch (Throwable ex) {
        LOGGER.error("Failed to deserialize data", ex);
      }
    }

    payload.release();

    return builder.build();
  }