Seek to current after exception; nested exception is org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method '**.eventhub.KafkaConsumer.consumeEventHubMessage(java.lang.String)' threw exception; nested exception is org.springframework.dao.InvalidDataAccessResourceUsageException: could not prepare statement; SQL [insert into d_ods_stream_portalone.tr_recepcao_portal_one (id_recepcao_portal_one, cnpjoper, nsu, cstatret, cuf, cam, camid, dat, dhpass, dhrespret, dhtransm, indiceconfianca, infleitura, latitude, longitude, nroeixos, pesobrutototal, plc, tpamb, tpambret, tpsentido, tptransm, velocidade, veraplic, veraplicret, version, xmotivoret) values (default, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)]; nested exception is org.hibernate.exception.SQLGrammarException: could not prepare statement; nested exception is org.springframework.dao.InvalidDataAccessResourceUsageException: could not prepare statement; SQL [insert into d_ods_stream_portalone.tr_recepcao_portal_one (id_recepcao_portal_one, cnpjoper, nsu, cstatret, cuf, cam, camid, dat, dhpass, dhrespret, dhtransm, indiceconfianca, infleitura, latitude, longitude, nroeixos, pesobrutototal, plc, tpamb, tpambret, tpsentido, tptransm, velocidade, veraplic, veraplicret, version, xmotivoret) values (default, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)]; nested exception is org.hibernate.exception.SQLGrammarException: could not prepare statement
Caused by: com.ibm.db2.jcc.am.SqlSyntaxErrorException: DB2 SQL Error: SQLCODE=-1667, SQLSTATE=42858, SQLERRMC=D_ODS_STREAM_PORTALONE.TR_RECEP;ORGANIZE BY COLUMN;FINAL|NEW|OLD TABLE, DRIVER=4.31.10
/**
* Recebe a mensagem do EventHub da Azure utilizando Kafka
*/
@Component
@Slf4j
public class KafkaConsumer {
private final PortalONERepository portalONERepository;
private final ObjectMapper objectMapper;
@Autowired
public KafkaConsumer(PortalONERepository portalONERepository, ObjectMapper objectMapper) {
this.portalONERepository = portalONERepository;
this.objectMapper = objectMapper;
}
@KafkaListener(topics = "divout")
@Transactional("transactionManager")
public void consumeEventHubMessage(String consumerMessage) {
log.info("Received message from kafka queue: {}", consumerMessage);
//Convert string message to java object
try {
DocumentONE[] documentOne = objectMapper.readValue(consumerMessage, DocumentONE[].class);
//Salvar cada mensagem no db2
portalONERepository.saveAll(Arrays.asList(documentOne));
} catch (JsonProcessingException e) {
log.error("Error receiving message: " + e.getMessage());
}
}
}
Arquivo properties:
spring:
application:
name: eventhub-stream
kafka:
bootstrap-servers: ***.servicebus.windows.net:9093
properties:
security:
protocol: SASL_SSL
sasl:
mechanism: PLAIN
jaas:
config: org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://**.servicebus.windows.net/;SharedAccessKeyName=***;SharedAccessKey=**=;EntityPath=**";
consumer:
group-id: doc-stream-consumer
key-deserialize: org.apache.kafka.common.serialization.StringSerializer
value-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
datasource:
name: URANO
url: jdbc:db2://****.br:50000/BLUDB
username: ${URANO_USER}
password: "${URANO_PASS}"
driver-class-name: com.ibm.db2.jcc.DB2Driver
hikari:
maximum-pool-size: 2
schema: D_ODS_STREAM_PORTALONE
jpa:
show-sql: true
generate-ddl: false
open-in-view: false
properties:
format_sql: true
hibernate:
ddl-auto: update
Classe model:
@Entity(name = "DOCUMENT_ONE")
@Table(name = "TR_RECEPCAO_PORTAL_ONE", schema = "D_ODS_STREAM_PORTALONE")
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
@NoArgsConstructor
@AllArgsConstructor
public class DocumentONE {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
@Column(name = "ID_RECEPCAO_PORTAL_ONE",nullable = false)
private BigInteger id;
private double version;
@Column(name = "TPAMB")
private byte tpAmb;
@Column(name = "VERAPLIC")
private String verAplic;
@Column(name = "TPTRANSM")
private char tpTransm;
@Column(name = "DHTRANSM")
@DateTimeFormat(iso = DateTimeFormat.ISO.TIME)
@JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd HH:mm:ss")
private LocalDateTime dhTransm;
@Column(name = "INFLEITURA")
private String infLeitura;
private String cUF;
@Column(name = "DHPASS")
@DateTimeFormat(iso = DateTimeFormat.ISO.TIME)
@JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd HH:mm:ss")
private LocalDateTime dhPass;
private BigInteger CNPJOper;
private String cam;
private String latitude;
private String longitude;
@DateTimeFormat(iso = DateTimeFormat.ISO.TIME)
@JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd HH:mm:ss")
private LocalDateTime dat;
@Column(name = "TPSENTIDO")
private char tpSentido;
private String plc;
private String camid;
private short velocidade;
@Column(name = "INDICECONFIANCA")
private short indiceConfianca;
@Column(name = "PESOBRUTOTOTAL")
private BigInteger pesoBrutoTotal;
@Column(name = "NROEIXOS")
private byte nroEixos;
@Column(name = "TPAMBRET")
private byte tpAmbRet;
@Column(name = "VERAPLICRET")
private String verAplicRet;
@Column(name = "CSTATRET")
private short cStatRet;
@Column(name = "XMOTIVORET")
private String xMotivoRet;
@Column(name = "DHRESPRET")
@DateTimeFormat(iso = DateTimeFormat.ISO.TIME)
@JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss-03:00")//"2022-11-06T01:56:09-03:00",
private LocalDateTime dhRespRet;
private BigInteger NSU;
}