SpringBoot集成Cassandra插入报错：UUID与String编解码器缺失

尝试通过SpringBoot向Cassandra数据库插入数据时遇到如下错误，参考数据类型映射文档后仍未解决：

reactor.core.Exceptions$ErrorCallbackNotImplemented:
com.datastax.oss.driver.api.core.type.codec.CodecNotFoundException:
Codec not found for requested operation: [UUID <-> java.lang.String]
Caused by:
com.datastax.oss.driver.api.core.type.codec.CodecNotFoundException:
Codec not found for requested operation: [UUID <-> java.lang.String]

实体类代码 #

package com.sams.pricing.prism.data.processor.entity;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.*;
import org.springframework.boot.autoconfigure.domain.EntityScan;
import org.springframework.data.cassandra.core.cql.PrimaryKeyType;
import org.springframework.data.cassandra.core.mapping.Column;
import org.springframework.data.cassandra.core.mapping.PrimaryKeyColumn;
import org.springframework.data.cassandra.core.mapping.Table;

import java.sql.Timestamp;

//import java.time.Instant;

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@JsonIgnoreProperties(ignoreUnknown = true)
@Table(value = "retail_lifecycle")
@EntityScan
@ToString
public class PricingLifeCycleEntity {

  @PrimaryKeyColumn(name = "itemnbr", ordinal = 0, type = PrimaryKeyType.PARTITIONED)
  private Integer itemNbr;

  @PrimaryKeyColumn(name = "clubnbr", ordinal = 1, type = PrimaryKeyType.CLUSTERED)
  private Short locationNbr;

  @PrimaryKeyColumn(name = "price", ordinal = 2, type = PrimaryKeyType.PARTITIONED)
  private Double price;

  @PrimaryKeyColumn(name = "pricestartts", ordinal = 3, type = PrimaryKeyType.CLUSTERED)
  private Timestamp priceStartTs;

  @PrimaryKeyColumn(name = "status", ordinal = 4, type = PrimaryKeyType.PARTITIONED)
  private String status;

  @Column(value = "createts")
  private Timestamp createTs;

  @Column(value = "currency")
  private String currency;
  
  @Column(value = "lastupdatets")
  private Timestamp lastUpdateTs;
  
  @Column(value = "priceendts")
  private Timestamp priceEndTs;
  
  @Column(value = "pricereasoncode")
  private String priceReasonCode;
  
  @Column(value = "pricereasoncodedesc")
  private String priceReasonCodeDesc;
  
  @Column(value = "pricetype")
  private String priceType;
  
  @Column(value = "pricetypedesc")
  private String priceTypeDesc;
   
  @Column(value = "statusdesc")
  private String statusDesc;
  
  @Column(value = "submissionclient")
  private String submissionClient;

  @Column(value = "submissionts")
  private Timestamp submissionTs;

  @Column(value = "submissionuserid")
  private String submissionUserId;
  
  @Column(value = "uniqueid")
  private String uniqueId;
 
}

数据库表结构 #

CREATE TABLE cassandra.prism.retail_lifecycle (
    itemnbr INTEGER NOT NULL,
    clubnbr SMALLINT NOT NULL,
    price DOUBLE NOT NULL,
    pricestartts TYPE_TIMESTAMP NOT NULL,
    status VARCHAR NOT NULL,
    createts TYPE_TIMESTAMP,
    currency VARCHAR,
    lastupdatets TYPE_TIMESTAMP,
    priceendts TYPE_TIMESTAMP,
    pricereasoncode VARCHAR,
    pricereasoncodedesc VARCHAR,
    pricetype VARCHAR,
    pricetypedesc VARCHAR,
    statusdesc VARCHAR,
    submissionclient VARCHAR,
    submissionts TYPE_TIMESTAMP,
    submissionuserid VARCHAR,
    uniqueid VARCHAR,
    CONSTRAINT RETAIL_LIFECYCLE_PK PRIMARY KEY (itemnbr,clubnbr,price,pricestartts,status)

测试用例代码 #

@Test
  public void test2() throws Exception {
    MockEndpoint mock =
        camelContext.getEndpoint("mock:" + Endpoints.SEDA_PROCESS_ENDPOINT, MockEndpoint.class);

    AdviceWith.adviceWith(
        camelContext,
        Endpoints.SEDA_SEND_ENDPOINT,
        // intercepting an exchange on route
        r -> {
          // replacing consumer with direct component
          r.replaceFromWith(Endpoints.SEDA_SEND_ENDPOINT);
          // mocking producer
          r.mockEndpoints("seda*");

          // weaving a last step to your route that will redirect the message to mock:result
          r.weaveAddLast().to(mock);
        });
    
    var eventId = UUID.randomUUID().toString();
    
    System.out.println("hereeeee "+eventId );
    String message = "{\"itemNbr\":8831,\"locationNbr\":4707,\"price\":55.0,\"priceStartTs\":\"2022-11-11T03:24:14.749Z\",\"status\":\"SUCCESS\",\"createTs\":\"2022-11-11T03:24:14.749Z\",\"currency\":\"USD\",\"lastUpdateTs\":\"2022-11-11T03:24:14.749Z\",\"priceEndTs\":\"2022-11-11T03:24:14.749Z\",\"priceReasonCode\":\"TEST\",\"priceReasonCodeDesc\":\"TEST\",\"priceType\":\"TST\",\"priceTypeDesc\":\"TST\",\"statusDesc\":\"TST\",\"submissionClient\":\"TST\",\"submissionTs\":\"2022-11-11T03:24:14.749Z\",\"submissionUserId\":\"TST\",\"uniqueId\":\"121212111212112121212121\"}";


    // setting expectations
    mock.expectedMessageCount(1);

    Map<String, Object> headers = new HashMap<>();
    headers.put(PrismConstants.APP_NAME, PrismConstants.PRICING_LIFE_CYCLE_APP_NAME);
    headers.put(PrismConstants.EVENT_ID, UUID.randomUUID().toString());

    // invoking consumer
    // we are going to send as a Json body object
    producerTemplate.sendBodyAndHeaders(
        Endpoints.SEDA_SEND_ENDPOINT,
        message,
        headers);

    // asserting mock is satisfied
    mock.assertIsSatisfied();
    var result = mock.getExchanges().get(0).getIn();
    System.out.println(result.getBody().toString());
  }

补充信息：更新后的主键实体类及数据库主键结构 #

主键实体类代码 #

package com.sams.pricing.prism.data.processor.entity;

import lombok.*;
import org.springframework.data.cassandra.core.cql.PrimaryKeyType;
import org.springframework.data.cassandra.core.mapping.PrimaryKeyClass;
import org.springframework.data.cassandra.core.mapping.PrimaryKeyColumn;

import java.sql.Timestamp;

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@PrimaryKeyClass
@ToString
public class PricingLifeCyclePrimaryKey {

    @PrimaryKeyColumn(name = "itemnbr", ordinal = 0, type = PrimaryKeyType.PARTITIONED)
    private Integer itemNumber;

    @PrimaryKeyColumn(name = "clubnbr", ordinal = 1, type = PrimaryKeyType.CLUSTERED)
    private Short locationNumber;

    @PrimaryKeyColumn(name = "price", ordinal = 2, type = PrimaryKeyType.PARTITIONED)
    private Double price;

    @PrimaryKeyColumn(name = "pricestartts", ordinal = 3, type = PrimaryKeyType.CLUSTERED)
    private Timestamp pricestartTs;

    @PrimaryKeyColumn(name = "status", ordinal = 4, type = PrimaryKeyType.PARTITIONED)
    private String status;


}

数据库主键结构截图 #

内容的提问来源于stack exchange，提问作者John