protected Object[] callRest( Object[] rowData ) throws KettleException {
    // get dynamic url ?
    if ( meta.isUrlInField() ) {
      data.realUrl = data.inputRowMeta.getString( rowData, data.indexOfUrlField );
    }
    // get dynamic method?
    if ( meta.isDynamicMethod() ) {
      data.method = data.inputRowMeta.getString( rowData, data.indexOfMethod );
      if ( Utils.isEmpty( data.method ) ) {
        throw new KettleException( BaseMessages.getString( PKG, "Rest.Error.MethodMissing" ) );
      }
    }
    WebTarget webResource = null;
    Client client = null;
    Object[] newRow = null;
    if ( rowData != null ) {
      newRow = rowData.clone();
    }
    try {
      if ( isDetailed() ) {
        logDetailed( BaseMessages.getString( PKG, "Rest.Log.ConnectingToURL", data.realUrl ) );
      }
      //      // Register a custom StringMessageBodyWriter to solve PDI-17423
      ClientBuilder clientBuilder = ClientBuilder.newBuilder();
      clientBuilder
        .withConfig( data.config )
        .property( HttpUrlConnectorProvider.SET_METHOD_WORKAROUND, true );
      if ( meta.isIgnoreSsl() || !Utils.isEmpty( data.trustStoreFile ) ) {
        clientBuilder.sslContext( data.sslContext );
        clientBuilder.hostnameVerifier( ( s1, s2 ) -> true );
      }
      client = clientBuilder.build();
      if ( data.basicAuthentication != null ) {
        client.register( data.basicAuthentication );
      }
      // create a WebResource object, which encapsulates a web resource for the client
      webResource = client.target( data.realUrl );

      // used for calculating the responseTime
      long startTime = System.currentTimeMillis();

      if ( data.useMatrixParams ) {
        // Add matrix parameters
        UriBuilder builder = webResource.getUriBuilder();
        for ( int i = 0; i < data.nrMatrixParams; i++ ) {
          String value = data.inputRowMeta.getString( rowData, data.indexOfMatrixParamFields[ i ] );
          if ( isDebug() ) {
            logDebug(
              BaseMessages.getString( PKG, "Rest.Log.matrixParameterValue", data.matrixParamNames[ i ], value ) );
          }
          builder = builder.matrixParam( data.matrixParamNames[ i ],
            UriComponent.encode( value, UriComponent.Type.QUERY_PARAM ) );
        }
        webResource = client.target( builder.build() );
      }

      if ( data.useParams ) {
        // Add query parameters
        for ( int i = 0; i < data.nrParams; i++ ) {
          String value = data.inputRowMeta.getString( rowData, data.indexOfParamFields[ i ] );
          if ( isDebug() ) {
            logDebug( BaseMessages.getString( PKG, "Rest.Log.queryParameterValue", data.paramNames[ i ], value ) );
          }
          webResource = webResource.queryParam( data.paramNames[ i ],  UriComponent.encode( value, UriComponent.Type.QUERY_PARAM ) );
        }
      }
      if ( isDebug() ) {
        logDebug( BaseMessages.getString( PKG, "Rest.Log.ConnectingToURL", webResource.getUri() ) );
      }
      Invocation.Builder invocationBuilder = webResource.request();
      String contentType = null; // media type override, if not null
      if ( data.useHeaders ) {
        // Add headers
        for ( int i = 0; i < data.nrheader; i++ ) {
          String value = data.inputRowMeta.getString( rowData, data.indexOfHeaderFields[ i ] );

          // unsure if an already set header will be returned to builder
          invocationBuilder.header( data.headerNames[ i ], value );
          if ( "Content-Type".equals( data.headerNames[ i ] ) ) {
            contentType = value;
          }
          if ( isDebug() ) {
            logDebug( BaseMessages.getString( PKG, "Rest.Log.HeaderValue", data.headerNames[ i ], value ) );
          }
        }
      }

      Response response;
      String entityString = "";
      if ( data.useBody ) {
        // Set Http request entity
        entityString = Const.NVL( data.inputRowMeta.getString( rowData, data.indexOfBodyField ), "" );
        if ( isDebug() ) {
          logDebug( BaseMessages.getString( PKG, "Rest.Log.BodyValue", entityString ) );
        }
      }
      try {
        if ( data.method.equals( RestMeta.HTTP_METHOD_GET ) ) {
          response = invocationBuilder.get( Response.class );
        } else if ( data.method.equals( RestMeta.HTTP_METHOD_POST ) ) {
          if ( null != contentType ) {
            response = invocationBuilder.post( Entity.entity( entityString, contentType ) );
          } else {
            //            response = builder.type( data.mediaType ).post( ClientResponse.class, entityString );
            response = invocationBuilder.post( Entity.entity( entityString, data.mediaType ) );
          }
        } else if ( data.method.equals( RestMeta.HTTP_METHOD_PUT ) ) {
          if ( null != contentType ) {
            response = invocationBuilder.put( Entity.entity( entityString, contentType ) );
          } else {
            response = invocationBuilder.put( Entity.entity( entityString, data.mediaType ) );
          }
        } else if ( data.method.equals( RestMeta.HTTP_METHOD_DELETE ) ) {
          response = invocationBuilder.delete();
        } else if ( data.method.equals( RestMeta.HTTP_METHOD_HEAD ) ) {
          response = invocationBuilder.head();
        } else if ( data.method.equals( RestMeta.HTTP_METHOD_OPTIONS ) ) {
          response = invocationBuilder.options();
        } else if ( data.method.equals( RestMeta.HTTP_METHOD_PATCH ) ) {
          if ( null != contentType ) {
            response =
              invocationBuilder.method(
                RestMeta.HTTP_METHOD_PATCH, Entity.entity( entityString, contentType ) );
          } else {
            response =
              invocationBuilder.method(
                RestMeta.HTTP_METHOD_PATCH, Entity.entity( entityString, data.mediaType ) );
          }
        } else {
          throw new KettleException( BaseMessages.getString( PKG, "Rest.Error.UnknownMethod", data.method ) );
        }
      } catch ( Exception e ) {
        throw new KettleException( "Request could not be processed", e );
      }
      // Get response time
      long responseTime = System.currentTimeMillis() - startTime;
      if ( isDetailed() ) {
        logDetailed(
          BaseMessages.getString( PKG, "Rest.Log.ResponseTime", String.valueOf( responseTime ), data.realUrl ) );
      }

      // Get status
      int status = response.getStatus();
      // Display status code
      if ( isDebug() ) {
        logDebug( BaseMessages.getString( PKG, "Rest.Log.ResponseCode", "" + status ) );
      }

      // Get Response
      String body;
      String headerString = null;
      try {
        body = response.readEntity( String.class );
      } catch ( Exception ex ) {
        body = "";
      }
      // get Header
      MultivaluedMap<String, Object> headers = searchForHeaders( response );
      JSONObject json = new JSONObject();
      for ( java.util.Map.Entry<String, List<Object>> entry : headers.entrySet() ) {
        String name = entry.getKey();
        List<Object> value = entry.getValue();
        if ( value.size() > 1 ) {
          json.put( name, value );
        } else {
          json.put( name, value.get( 0 ) );
        }
      }
      headerString = json.toJSONString();
      // for output
      int returnFieldsOffset = data.inputRowMeta.size();
      // add response to output
      if ( !Utils.isEmpty( data.resultFieldName ) ) {
        newRow = RowDataUtil.addValueData( newRow, returnFieldsOffset, body );
        returnFieldsOffset++;
      }

      // add status to output
      if ( !Utils.isEmpty( data.resultCodeFieldName ) ) {
        newRow = RowDataUtil.addValueData( newRow, returnFieldsOffset, new Long( status ) );
        returnFieldsOffset++;
      }

      // add response time to output
      if ( !Utils.isEmpty( data.resultResponseFieldName ) ) {
        newRow = RowDataUtil.addValueData( newRow, returnFieldsOffset, new Long( responseTime ) );
        returnFieldsOffset++;
      }
      // add response header to output
      if ( !Utils.isEmpty( data.resultHeaderFieldName ) ) {
        newRow = RowDataUtil.addValueData( newRow, returnFieldsOffset, headerString );
      }
    } catch ( Exception e ) {
      throw new KettleException( BaseMessages.getString( PKG, "Rest.Error.CanNotReadURL", data.realUrl ), e );
    } finally {
      if ( webResource != null ) {
        webResource = null;
      }
      if ( client != null ) {
        client.close();
      }
    }
    return newRow;
  }
@Test
  public void testPutWithEmptyBody() throws KettleException {

    Invocation.Builder builder = mock( Invocation.Builder.class );

    WebTarget resource = mock( WebTarget.class );
    lenient().doReturn( builder ).when( resource ).request();

    Client client = mock( Client.class );
    lenient().doReturn( resource ).when( client ).target( anyString() );

    ClientBuilder clientBuilder = mock( ClientBuilder.class );
    lenient().when( clientBuilder.build() ).thenReturn( client );

    RestMeta meta = mock( RestMeta.class );
    lenient().doReturn( false ).when( meta ).isUrlInField();
    lenient().doReturn( false ).when( meta ).isDynamicMethod();

    RowMetaInterface rmi = mock( RowMetaInterface.class );

    RestData data = mock( RestData.class );
    data.method = RestMeta.HTTP_METHOD_PUT;
    data.config = new ClientConfig();
    data.inputRowMeta = rmi;
    data.realUrl = "http://localhost:8080/pentaho";
    data.mediaType = MediaType.TEXT_PLAIN_TYPE;
    data.useBody = true;
    // do not set data.indexOfBodyField

    Rest rest = mock( Rest.class );
    doCallRealMethod().when( rest ).callRest( any() );

    ReflectionTestUtils.setField( rest, "meta", meta );
    ReflectionTestUtils.setField( rest, "data", data );

    try {
      rest.callRest( new Object[] { 0 } );
    } catch ( Exception exception ) {
      // Ignore the ConnectException which is expected as rest call to localhost:8080 will fail in unit test
      // IllegalStateException is throws when the body is null
      if ( exception.getCause().getCause() instanceof IllegalStateException ) {
          Assert.fail( "PUT request with an empty body should not have failed with an IllegalStateException" );
      }
    }
  }