CODEWITHSUNDEEP
spring-cloud-streamspring-kafka
sunsets
sunsets

Reputation: 411

spring kafka, MessageDeliveryException: failed to send Message to channel

I'm struggling with the problem about a source, a process, and sink. I have two project, application. One is for the source and the other one is for the process and sink.

Let me share each component to figure out what happened.

@EnableBinding(MultiProducerChannel.class)
public class RealTimeDataSource{

@Autowired
RealTimeProductionService realTimeProductionService;

@InboundChannelAdapter(value = MultiProducerChannel.SOURCEPRODUCTION, poller = @Poller(fixedDelay = "10000", maxMessagesPerPoll = "1"))
public JSONArray productionMessageSource() throws Exception {

    long currentTime = System.currentTimeMillis();

    JSONArray realTimeProductionList = realTimeProductionService.getNewProductionTime();

    System.out.println(currentTime + " : Running source...");
    return realTimeProductionList;

    }

}


@EnableBinding(MultiChannel.class)
public class RealTimeDataProcessor {

@Autowired
RealTimeProductionService realTimeProductionService;

@Transformer(inputChannel = MultiChannel.PROCESSPRODUCTION, outputChannel = MultiChannel.SAVEPRODUCTION)
public JSONObject productionMessageProcessor(List<RealTimeProduction> realTimeProductionList) throws Exception {

    JSONObject jsonObject = null;
    if(realTimeProductionList != null) {
        jsonObject = new JSONObject(realTimeProductionService.getNewProductionTime(realTimeProductionList));
        System.out.println("PROCESSOR RUNNING...");
    }

    return jsonObject;
    }
}

@EnableBinding(MultiChannel.class)
public class RealTimeDataSink {

private static final String INDEX_NAME = "c000001_kr_50879_f01";


@Autowired
private JestClient jestClient;

@StreamListener(MultiChannel.SAVEFINALPRODUCTION)
public void productionMessageSink(JSONObject outputs) throws Exception {

    if (outputs != null) {

        boolean indexExists = jestClient.execute(new IndicesExists.Builder(INDEX_NAME).build()).isSucceeded();

        JestResult jestResult = jestClient.execute(new Index.Builder(outputs).index(INDEX_NAME).type("production").build());




        System.out.println("SINK RUNNING...");

    }
}

}

   public interface MultiChannel {

String SOURCEPRODUCTION = "production-source";

String PROCESSPRODUCTION = "production-process";

String SAVEPRODUCTION = "production-save";

String SAVEFINALPRODUCTION = "production-save-final";

@Output(SOURCEPRODUCTION)
MessageChannel sourceproduction();

@Input(PROCESSPRODUCTION)
SubscribableChannel processorproduction();

@Output(SAVEPRODUCTION)
MessageChannel saveproduction();

@Input(SAVEFINALPRODUCTION)
SubscribableChannel savefinalproduction();

}

I think the souce is working well. But I don't catch any idea to find the problem about this error. I spend three whole days.... still... doesn't worrk.

org.springframework.messaging.MessageDeliveryException: failed to send Message to channel 'production-save'; nested exception is java.lang.IllegalArgumentException: payload must not be null

any idea about that?

Upvotes: 0

Views: 3173

Answers (1)

Artem Bilan
Artem Bilan

Reputation: 121177

Consider to use @StreamListener and @SendTo instead of @Transformer for automatic content-type handling: https://docs.spring.io/spring-cloud-stream/docs/Ditmars.RC1/reference/htmlsingle/#_using_streamlistener_for_automatic_content_type_handling.

If that doesn't help consider to configure contentType: application/json for the production-save binding: https://docs.spring.io/spring-cloud-stream/docs/Ditmars.RC1/reference/htmlsingle/#_properties_for_use_of_spring_cloud_stream

Upvotes: 1

Related Questions