private MqttMessage getMessage( Object[] row ) throws KettleStepException {
    MqttMessage mqttMessage = new MqttMessage();
    try {
      mqttMessage.setQos( Integer.parseInt( meta.qos ) );
    } catch ( NumberFormatException e ) {
      throw new KettleStepException(
        getString( PKG, "MQTTProducer.Error.QOS", meta.qos ) );
    }
    //noinspection ConstantConditions
    mqttMessage.setPayload( getFieldData( row, meta.messageField )
      .map( this::dataAsBytes )
      .orElse( null ) ); //allow nulls to pass through
    return mqttMessage;
  }
@Test
  public void testErrorOnPublishStopsAll() throws Exception {
    handleAsSecondRow( trans );

    MqttException mqttException = mock( MqttException.class );
    when( mqttException.getMessage() ).thenReturn( "publish failed" );
    when( mqttClient.isConnected() ).thenReturn( true, false );
    doThrow( mqttException ).when( mqttClient ).publish( any(), any() );

    trans.startThreads();
    trans.waitUntilFinished();

    verify( mqttClient ).disconnect();
    verify( logChannel ).logError(
      "MQTT Producer - Received an exception publishing the message."
        + "  Check that Quality of Service level 0 is supported by your MQTT Broker" );
    verify( logChannel ).logError( "publish failed", mqttException );
    assertEquals( 0, trans.getSteps().get( 1 ).step.getLinesOutput() );
  }