Not able to publish and consume message with spring-cloud-stream-test-binder:4.0.4

Question

I am using spring-cloud-stream-test-binder with version 4.0.4 But while publishing/consuming i am getting below error

java.lang.NullPointerException: Cannot invoke "org.springframework.messaging.SubscribableChannel.send(org.springframework.messaging.Message)" because the return value of "org.springframework.cloud.stream.binder.test.InputDestination.getChannelByName(String)" is null at org.springframework.cloud.stream.binder.test.InputDestination.send(InputDestination.java:89) at com.csn.tax.pfd.w2.evt.subscriber.service.ClientDataSubscriberTests.testExceptionOnGenerateW2Data(ClientDataSubscriberTests.java:512) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.base/java.lang.reflect.Method.invoke(Method.java:568)

My Code Snippet: Code Snippet Of Junit:

@DirtiesContext
@SpringBootTest
@AutoConfigureMockMvc
@TestPropertySource("classpath:test.properties")
@ImportAutoConfiguration(
    classes = TestChannelBinderConfiguration.class,
    exclude = {
            KafkaAutoConfiguration.class,
            KafkaMetricsAutoConfiguration.class,
            DataSourceAutoConfiguration.class,
            TransactionAutoConfiguration.class,
            DataSourceTransactionManagerAutoConfiguration.class})
public class ClientDataSubscriberTests {
    
static final String INPUT_TOPIC_NAME = "tax-datafeed-evt-load-clt-dnr";

static final String OUTPUT_TOPIC_NAME = "tax-statusdata-evt-load-clt-dnr";
@Test
public void testExceptionOnGenerateW2Data() {
    String fileId = "testFileId";
    String loadId = "testLoadId";
    String runId = "testRunId";
    String branchNumber = "0080";
    String clientAccountNumber = "INVALID_CLIENT";
    String quarter = "3";
    String year = "2020";

    // Build Kafka event with INVALID_CLIENT
    PayrollClientData payrollClientData = PayrollClientData.newBuilder()
            .setFileId(fileId)
            .setLoadId(loadId)
            .setRunId(runId)
            .setBranchNumber(branchNumber)
            .setClientAccountId(clientAccountNumber)
            .setQuarter(quarter)
            .setYear(year)
            .build();

    Message<PayrollClientData> clientDataMessage = MessageBuilder.withPayload(payrollClientData).build();

    // Send input message
    inputDestination.send(clientDataMessage, INPUT_TOPIC_NAME);

    // Receive output message
    @SuppressWarnings("unchecked")
    Message<PayrollLoadStatusData> payrollLoadStatusMessage = (Message<PayrollLoadStatusData>) (Object)
            outputDestination.receive(500, OUTPUT_TOPIC_NAME);

    PayrollLoadStatusData receivedPayrollLoadStatusData = payrollLoadStatusMessage.getPayload();

    // Assert that received Kafka event contains the expected data for each field
    assertNotNull(receivedPayrollLoadStatusData);

    assertEquals(receivedPayrollLoadStatusData.getRunId().toString(), runId);
    assertEquals(receivedPayrollLoadStatusData.getLoadId().toString(), loadId);
    assertEquals(receivedPayrollLoadStatusData.getFileId().toString(), fileId);
    assertEquals(receivedPayrollLoadStatusData.getQuarter().toString(), quarter);
    assertEquals(receivedPayrollLoadStatusData.getYear().toString(), year);
    assertEquals(receivedPayrollLoadStatusData.getClientAccountId().toString(), clientAccountNumber);
    assertEquals(receivedPayrollLoadStatusData.getBranchNumber().toString(), branchNumber);
    assertEquals(receivedPayrollLoadStatusData.getStatusType(), PayrollLoadStatusType.FAILED);
    
}
}


**Code snippet Of Main Class :**

@SpringBootApplication(scanBasePackages = 
{"com.csn.evt.subscriber"},
                   exclude = {DataSourceAutoConfiguration.class,
                           MongoAutoConfiguration.class,
                           MongoDataAutoConfiguration.class})
@EnableRetry
@EnableMongoAuditing
@EnableAsync
@EnableScheduling
public class Application {
public static void main(String[] args) {
    SpringApplication.run(Application.class, args);
}
}

score 0 · Answer 1 · answered Aug 08 '23 at 11:41

0

The code snippet you had posted is not helpful, since it does not show anything about how your test is setup. Consider looking at some of our tests where we use Test Binder for an example https://github.com/spring-cloud/spring-cloud-stream/blob/main/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java. Also, you can see an isolated sample here

answered Aug 08 '23 at 11:41

Oleg Zhurakousky

5,820
16
17

I tried as per above URLs but i am getting error with fetching values for properties mentioned inside "@TestPropertySource("classpath:test.properties")". So as per your above comment i have added code to get more idea about my Test set up and application set up . Now could you please guide how to implement my junit test @OlegZhurakousky – sekhar Aug 09 '23 at 13:17
I tried as per above URLs but i am getting error with fetching values for properties mentioned inside "@TestPropertySource("classpath:test.properties")". So as per your above comment i have added code to get more idea about my Test set up and application set up . Now could you please guide how to implement my junit test @Oleg Zhurakousky – sekhar Aug 09 '23 at 13:41
I already did guide you with working sample, our existing test cases and documentation that provides yet another working sample. Not sure what else I can do – Oleg Zhurakousky Aug 14 '23 at 07:18

Not able to publish and consume message with spring-cloud-stream-test-binder:4.0.4

1 Answers1