alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-8164 Upgrade Outbox Extension to Quarkus 3.14.0

Browse Source
This commit is contained in:
Chris Cranford 2024-08-21 19:14:21 -04:00 committed by Jiri Pechanec
parent 97c4ccbd52
commit f3f8621e97
12 changed files with 114 additions and 129 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

131
debezium-quarkus-outbox-common/deployment/src/main/java/io/debezium/outbox/quarkus/deployment/DebeziumOutboxCommonConfig.java
View File

@ -8,10 +8,11 @@
import java.util.Optional;
import io.debezium.outbox.quarkus.internal.OutboxConstants;
import io.quarkus.runtime.annotations.ConfigGroup;
import io.quarkus.runtime.annotations.ConfigItem;
import io.quarkus.runtime.annotations.ConfigPhase;
import io.quarkus.runtime.annotations.ConfigRoot;
import io.smallrye.config.ConfigMapping;
import io.smallrye.config.WithDefault;
import io.smallrye.config.WithName;
/**
* Configuration root class for Debezium Outbox pattern that defines the available user
@ -19,211 +20,185 @@
*
* @author Chris Cranford
*/
@ConfigRoot(phase = ConfigPhase.BUILD_TIME, name = "debezium-outbox")
public class DebeziumOutboxCommonConfig {
@ConfigMapping(prefix = "quarkus.debezium-outbox")
@ConfigRoot(phase = ConfigPhase.BUILD_TIME)
public interface DebeziumOutboxCommonConfig {
/**
* The table name to be used for the outbox table
*/
@ConfigItem(defaultValue = "OutboxEvent")
public String tableName;
@WithDefault("OutboxEvent")
String tableName();
/**
* Outbox identifier configurable attributes
*/
@ConfigItem
public DebeziumOutboxConfigId id;
DebeziumOutboxConfigId id();
/**
* Outbox aggregate-id configurable attributes
*/
@ConfigItem
public DebeziumOutboxConfigAggregateId aggregateId;
DebeziumOutboxConfigAggregateId aggregateId();
/**
* Outbox aggregate-type configurable attributes
*/
@ConfigItem
public DebeziumOutboxConfigAggregateType aggregateType;
DebeziumOutboxConfigAggregateType aggregateType();
/**
* Outbox type configurable attributes
*/
@ConfigItem
public DebeziumOutboxConfigType type;
DebeziumOutboxConfigType type();
/**
* Outbox timestamp configurable attributes
*/
@ConfigItem
public DebeziumOutboxConfigTimestamp timestamp;
DebeziumOutboxConfigTimestamp timestamp();
/**
* Outbox payload configurable attributes
*/
@ConfigItem
public DebeziumOutboxConfigPayload payload;
DebeziumOutboxConfigPayload payload();
/**
* Outbox additional fields
*/
@ConfigItem
public Optional<String> additionalFields;
Optional<String> additionalFields();
/**
* Outbox Tracing configurable attributes
*/
@ConfigItem
public DebeziumOutboxConfigTracingSpan tracingSpan;
DebeziumOutboxConfigTracingSpan tracingSpan();
/**
* OpenTelemetry configuration option
*/
@ConfigItem(name = "tracing.enabled", defaultValue = "true")
public boolean tracingEnabled;
@WithDefault("true")
@WithName("tracing.enabled")
boolean tracingEnabled();
@ConfigGroup
public static class DebeziumOutboxConfigId {
interface DebeziumOutboxConfigId {
/**
* The column name.
*/
@ConfigItem(defaultValue = "id")
public String name;
@WithDefault("id")
String name();
/**
* The column definition.
*/
@ConfigItem
public Optional<String> columnDefinition;
Optional<String> columnDefinition();
}
@ConfigGroup
public static class DebeziumOutboxConfigAggregateType {
interface DebeziumOutboxConfigAggregateType {
/**
* The column name.
*/
@ConfigItem(defaultValue = "aggregatetype")
public String name;
@WithDefault("aggregatetype")
String name();
/**
* The column definition.
*/
@ConfigItem
public Optional<String> columnDefinition;
Optional<String> columnDefinition();
/**
* The column's attribute converter fully qualified class name.
* @see jakarta.persistence.AttributeConverter
*/
@ConfigItem
public Optional<String> converter;
Optional<String> converter();
}
@ConfigGroup
public static class DebeziumOutboxConfigAggregateId {
interface DebeziumOutboxConfigAggregateId {
/**
* The column name.
*/
@ConfigItem(defaultValue = "aggregateid")
public String name;
@WithDefault("aggregateid")
String name();
/**
* The column definition.
*/
@ConfigItem
public Optional<String> columnDefinition;
Optional<String> columnDefinition();
/**
* The column's attribute converter fully qualified class name.
* @see jakarta.persistence.AttributeConverter
*/
@ConfigItem
public Optional<String> converter;
Optional<String> converter();
}
@ConfigGroup
public static class DebeziumOutboxConfigType {
interface DebeziumOutboxConfigType {
/**
* The column name.
*/
@ConfigItem(defaultValue = "type")
public String name;
@WithDefault("type")
String name();
/**
* The column definition.
*/
@ConfigItem
public Optional<String> columnDefinition;
Optional<String> columnDefinition();
/**
* The column's attribute converter fully qualified class name.
* @see jakarta.persistence.AttributeConverter
*/
@ConfigItem
public Optional<String> converter;
Optional<String> converter();
}
@ConfigGroup
public static class DebeziumOutboxConfigTimestamp {
interface DebeziumOutboxConfigTimestamp {
/**
* The column name.
*/
@ConfigItem(defaultValue = "timestamp")
public String name;
@WithDefault("timestamp")
String name();
/**
* The column definition.
*/
@ConfigItem
public Optional<String> columnDefinition;
Optional<String> columnDefinition();
/**
* The column's attribute converter fully qualified class name.
* @see jakarta.persistence.AttributeConverter
*/
@ConfigItem
public Optional<String> converter;
Optional<String> converter();
}
@ConfigGroup
public static class DebeziumOutboxConfigPayload {
interface DebeziumOutboxConfigPayload {
/**
* The column name.
*/
@ConfigItem(defaultValue = "payload")
public String name;
@WithDefault("payload")
String name();
/**
* The column definition.
*/
@ConfigItem
public Optional<String> columnDefinition;
Optional<String> columnDefinition();
/**
* The column's attribute converter fully qualified class name.
* @see jakarta.persistence.AttributeConverter
*/
@ConfigItem
public Optional<String> converter;
Optional<String> converter();
/**
* The column's type definition class
*/
@ConfigItem
public Optional<String> type;
Optional<String> type();
}
@ConfigGroup
public static class DebeziumOutboxConfigTracingSpan {
interface DebeziumOutboxConfigTracingSpan {
/**
* The column name.
*/
@ConfigItem(defaultValue = OutboxConstants.TRACING_SPAN_CONTEXT)
public String name;
@WithDefault(OutboxConstants.TRACING_SPAN_CONTEXT)
String name();
/**
* The column definition.
*/
@ConfigItem
public Optional<String> columnDefinition;
Optional<String> columnDefinition();
}
}

66
debezium-quarkus-outbox-common/deployment/src/main/java/io/debezium/outbox/quarkus/deployment/OutboxEventHbmWriter.java
View File

@ -39,7 +39,7 @@ static JaxbHbmHibernateMapping write(DebeziumOutboxCommonConfig config, OutboxEv
final JaxbHbmRootEntityType entityType = new JaxbHbmRootEntityType();
entityType.setEntityName(OUTBOX_ENTITY_FULLNAME);
entityType.setTable(config.tableName);
entityType.setTable(config.tableName());
mapping.getClazz().add(entityType);
// Setup generator
@ -55,13 +55,13 @@ static JaxbHbmHibernateMapping write(DebeziumOutboxCommonConfig config, OutboxEv
entityType.getAttributes().add(createTypeAttribute(config));
entityType.getAttributes().add(createTimestampAttribute(config));
entityType.getAttributes().add(createPayloadAttribute(config, outboxEventEntityBuildItem));
if (config.tracingEnabled) {
if (config.tracingEnabled()) {
entityType.getAttributes().add(createTracingSpanAttribute(config));
}
// Additional fields
if (config.additionalFields.isPresent()) {
String[] fields = config.additionalFields.get().split(",");
if (config.additionalFields().isPresent()) {
String[] fields = config.additionalFields().get().split(",");
for (int fieldIndex = 0; fieldIndex < fields.length; ++fieldIndex) {
String[] parts = fields[fieldIndex].split(":");
if (parts.length < 2) {
@ -96,8 +96,8 @@ private static JaxbHbmSimpleIdType createIdAttribute(DebeziumOutboxCommonConfig
attribute.setTypeAttribute(UUID.class.getName());
final JaxbHbmColumnType column = new JaxbHbmColumnType();
column.setName(config.id.name);
config.id.columnDefinition.ifPresent(column::setSqlType);
column.setName(config.id().name());
config.id().columnDefinition().ifPresent(column::setSqlType);
attribute.getColumn().add(column);
final JaxbHbmGeneratorSpecificationType generator = new JaxbHbmGeneratorSpecificationType();
@ -111,16 +111,16 @@ private static JaxbHbmBasicAttributeType createAggregateTypeAttribute(DebeziumOu
final JaxbHbmBasicAttributeType attribute = new JaxbHbmBasicAttributeType();
attribute.setName("aggregateType");
attribute.setNotNull(true);
if (config.aggregateType.converter.isPresent()) {
attribute.setTypeAttribute("converted::" + config.aggregateType.converter.get());
if (config.aggregateType().converter().isPresent()) {
attribute.setTypeAttribute("converted::" + config.aggregateType().converter().get());
}
else {
attribute.setTypeAttribute("string");
}
final JaxbHbmColumnType column = new JaxbHbmColumnType();
column.setName(config.aggregateType.name);
config.aggregateType.columnDefinition.ifPresent(column::setSqlType);
column.setName(config.aggregateType().name());
config.aggregateType().columnDefinition().ifPresent(column::setSqlType);
attribute.getColumnOrFormula().add(column);
return attribute;
@ -131,16 +131,16 @@ private static JaxbHbmBasicAttributeType createAggregateIdAttribute(DebeziumOutb
final JaxbHbmBasicAttributeType attribute = new JaxbHbmBasicAttributeType();
attribute.setName("aggregateId");
attribute.setNotNull(true);
if (config.aggregateId.converter.isPresent()) {
attribute.setTypeAttribute("converted::" + config.aggregateId.converter.get());
if (config.aggregateId().converter().isPresent()) {
attribute.setTypeAttribute("converted::" + config.aggregateId().converter().get());
}
else {
attribute.setTypeAttribute(outboxEventEntityBuildItem.getAggregateIdType().name().toString());
}
final JaxbHbmColumnType column = new JaxbHbmColumnType();
column.setName(config.aggregateId.name);
config.aggregateId.columnDefinition.ifPresent(column::setSqlType);
column.setName(config.aggregateId().name());
config.aggregateId().columnDefinition().ifPresent(column::setSqlType);
attribute.getColumnOrFormula().add(column);
return attribute;
@ -150,16 +150,16 @@ private static JaxbHbmBasicAttributeType createTypeAttribute(DebeziumOutboxCommo
final JaxbHbmBasicAttributeType attribute = new JaxbHbmBasicAttributeType();
attribute.setName("type");
attribute.setNotNull(true);
if (config.type.converter.isPresent()) {
attribute.setTypeAttribute("converted::" + config.type.converter.get());
if (config.type().converter().isPresent()) {
attribute.setTypeAttribute("converted::" + config.type().converter().get());
}
else {
attribute.setTypeAttribute("string");
}
final JaxbHbmColumnType column = new JaxbHbmColumnType();
column.setName(config.type.name);
config.type.columnDefinition.ifPresent(column::setSqlType);
column.setName(config.type().name());
config.type().columnDefinition().ifPresent(column::setSqlType);
attribute.getColumnOrFormula().add(column);
return attribute;
@ -169,8 +169,8 @@ private static JaxbHbmBasicAttributeType createTimestampAttribute(DebeziumOutbox
final JaxbHbmBasicAttributeType attribute = new JaxbHbmBasicAttributeType();
attribute.setName("timestamp");
attribute.setNotNull(true);
if (config.timestamp.converter.isPresent()) {
attribute.setTypeAttribute("converted::" + config.timestamp.converter.get());
if (config.timestamp().converter().isPresent()) {
attribute.setTypeAttribute("converted::" + config.timestamp().converter().get());
}
else {
// Hibernate 6.x expects full qualified class names for this whereas Hibernate 5.x had
@ -179,8 +179,8 @@ private static JaxbHbmBasicAttributeType createTimestampAttribute(DebeziumOutbox
}
final JaxbHbmColumnType column = new JaxbHbmColumnType();
column.setName(config.timestamp.name);
config.timestamp.columnDefinition.ifPresent(column::setSqlType);
column.setName(config.timestamp().name());
config.timestamp().columnDefinition().ifPresent(column::setSqlType);
attribute.getColumnOrFormula().add(column);
return attribute;
@ -195,13 +195,13 @@ private static JaxbHbmBasicAttributeType createPayloadAttribute(DebeziumOutboxCo
attribute.setName("payload");
attribute.setNotNull(false);
if (config.payload.type.isPresent()) {
LOGGER.info("Using payload type: {}", config.payload.type.get());
attribute.setTypeAttribute(config.payload.type.get());
if (config.payload().type().isPresent()) {
LOGGER.info("Using payload type: {}", config.payload().type().get());
attribute.setTypeAttribute(config.payload().type().get());
}
else if (config.payload.converter.isPresent()) {
LOGGER.info("Using payload attribute converter: {}", config.payload.converter.get());
attribute.setTypeAttribute("converted::" + config.payload.converter.get());
else if (config.payload().converter().isPresent()) {
LOGGER.info("Using payload attribute converter: {}", config.payload().converter().get());
attribute.setTypeAttribute("converted::" + config.payload().converter().get());
}
else if (isJacksonJsonNode) {
LOGGER.info("Using payload attribute converter: {}", JsonNodeAttributeConverter.class.getName());
@ -214,10 +214,10 @@ else if (isJacksonJsonNode) {
}
final JaxbHbmColumnType column = new JaxbHbmColumnType();
column.setName(config.payload.name);
column.setName(config.payload().name());
if (config.payload.columnDefinition.isPresent()) {
column.setSqlType(config.payload.columnDefinition.get());
if (config.payload().columnDefinition().isPresent()) {
column.setSqlType(config.payload().columnDefinition().get());
}
else if (isJacksonJsonNode) {
column.setSqlType("varchar(8000)");
@ -235,10 +235,10 @@ private static JaxbHbmBasicAttributeType createTracingSpanAttribute(DebeziumOutb
attribute.setTypeAttribute("string");
final JaxbHbmColumnType column = new JaxbHbmColumnType();
column.setName(config.tracingSpan.name);
column.setName(config.tracingSpan().name());
column.setLength(256);
config.tracingSpan.columnDefinition.ifPresent(column::setSqlType);
config.tracingSpan().columnDefinition().ifPresent(column::setSqlType);
attribute.getColumnOrFormula().add(column);

12
debezium-quarkus-outbox-common/runtime/src/main/java/io/debezium/outbox/quarkus/internal/DebeziumOutboxCommonRuntimeConfig.java
View File

@ -5,20 +5,22 @@
*/
package io.debezium.outbox.quarkus.internal;
import io.quarkus.runtime.annotations.ConfigItem;
import io.quarkus.runtime.annotations.ConfigPhase;
import io.quarkus.runtime.annotations.ConfigRoot;
import io.smallrye.config.ConfigMapping;
import io.smallrye.config.WithDefault;
/**
* Debezium outbox Quarkus extension common runtime configuration properties.
*
* @author Chris Cranford
*/
@ConfigRoot(phase = ConfigPhase.RUN_TIME, name = "debezium-outbox")
public class DebeziumOutboxCommonRuntimeConfig {
@ConfigMapping(prefix = "quarkus.debezium-outbox")
@ConfigRoot(phase = ConfigPhase.RUN_TIME)
public interface DebeziumOutboxCommonRuntimeConfig {
/**
* Remove outbox entity after being inserted. Default is {@code true}.
*/
@ConfigItem(defaultValue = "true")
public boolean removeAfterInsert;
@WithDefault("true")
boolean removeAfterInsert();
}

6
debezium-quarkus-outbox-reactive/deployment/src/main/java/io/debezium/outbox/reactive/quarkus/deployment/DebeziumOutboxConfig.java
View File

@ -8,6 +8,7 @@
import io.debezium.outbox.quarkus.deployment.DebeziumOutboxCommonConfig;
import io.quarkus.runtime.annotations.ConfigPhase;
import io.quarkus.runtime.annotations.ConfigRoot;
import io.smallrye.config.ConfigMapping;
/**
* Configuration root class for Debezium Outbox pattern that defines the available user
@ -15,7 +16,8 @@
*
* @author Chris Cranford
*/
@ConfigRoot(phase = ConfigPhase.BUILD_TIME, name = "debezium-outbox")
public class DebeziumOutboxConfig extends DebeziumOutboxCommonConfig {
@ConfigMapping(prefix = "quarkus.debezium-outbox")
@ConfigRoot(phase = ConfigPhase.BUILD_TIME)
public interface DebeziumOutboxConfig extends DebeziumOutboxCommonConfig {
}

2
debezium-quarkus-outbox-reactive/deployment/src/main/java/io/debezium/outbox/reactive/quarkus/deployment/OutboxProcessor.java
View File

@ -42,7 +42,7 @@ public void build(OutboxEventEntityBuildItem outboxBuildItem,
BuildProducer<GeneratedResourceBuildItem> generatedResourcesProducer,
BuildProducer<ReflectiveClassBuildItem> reflectiveClassProducer,
Capabilities capabilities) {
if (debeziumOutboxConfig.tracingEnabled && capabilities.isPresent(Capability.OPENTELEMETRY_TRACER)) {
if (debeziumOutboxConfig.tracingEnabled() && capabilities.isPresent(Capability.OPENTELEMETRY_TRACER)) {
additionalBeanProducer.produce(AdditionalBeanBuildItem.unremovableOf(DebeziumTracerEventDispatcher.class));
}

2
debezium-quarkus-outbox-reactive/runtime/src/main/java/io/debezium/outbox/reactive/quarkus/internal/AbstractEventDispatcher.java
View File

@ -48,7 +48,7 @@ protected Uni<Void> persist(Map<String, Object> dataMap) {
}
protected Uni<Integer> removeFromOutbox(Map<String, Object> dataMap) {
if (config.removeAfterInsert) {
if (config.removeAfterInsert()) {
LOGGER.debug("removing outbox event");
return factory.withSession(
session -> session.withTransaction(

6
debezium-quarkus-outbox-reactive/runtime/src/main/java/io/debezium/outbox/reactive/quarkus/internal/DebeziumOutboxRuntimeConfig.java
View File

@ -8,13 +8,15 @@
import io.debezium.outbox.quarkus.internal.DebeziumOutboxCommonRuntimeConfig;
import io.quarkus.runtime.annotations.ConfigPhase;
import io.quarkus.runtime.annotations.ConfigRoot;
import io.smallrye.config.ConfigMapping;
/**
* Debezium outbox Quarkus extension runtime configuration properties.
*
* @author Chris Cranford
*/
@ConfigRoot(phase = ConfigPhase.RUN_TIME, name = "debezium-outbox")
public class DebeziumOutboxRuntimeConfig extends DebeziumOutboxCommonRuntimeConfig {
@ConfigMapping(prefix = "quarkus.debezium-outbox")
@ConfigRoot(phase = ConfigPhase.RUN_TIME)
public interface DebeziumOutboxRuntimeConfig extends DebeziumOutboxCommonRuntimeConfig {
}

6
debezium-quarkus-outbox/deployment/src/main/java/io/debezium/outbox/quarkus/deployment/DebeziumOutboxConfig.java
View File

@ -7,6 +7,7 @@
import io.quarkus.runtime.annotations.ConfigPhase;
import io.quarkus.runtime.annotations.ConfigRoot;
import io.smallrye.config.ConfigMapping;
/**
* Configuration root class for Debezium Outbox pattern that defines the available user
@ -14,7 +15,8 @@
*
* @author Chris Cranford
*/
@ConfigRoot(phase = ConfigPhase.BUILD_TIME, name = "debezium-outbox")
public class DebeziumOutboxConfig extends DebeziumOutboxCommonConfig {
@ConfigMapping(prefix = "quarkus.debezium-outbox")
@ConfigRoot(phase = ConfigPhase.BUILD_TIME)
public interface DebeziumOutboxConfig extends DebeziumOutboxCommonConfig {
}

2
debezium-quarkus-outbox/deployment/src/main/java/io/debezium/outbox/quarkus/deployment/OutboxProcessor.java
View File

@ -38,7 +38,7 @@ public void build(OutboxEventEntityBuildItem outboxBuildItem,
BuildProducer<GeneratedResourceBuildItem> generatedResourcesProducer,
BuildProducer<ReflectiveClassBuildItem> reflectiveClassProducer,
Capabilities capabilities) {
if (getConfig().tracingEnabled && capabilities.isPresent(Capability.OPENTELEMETRY_TRACER)) {
if (getConfig().tracingEnabled() && capabilities.isPresent(Capability.OPENTELEMETRY_TRACER)) {
additionalBeanProducer.produce(AdditionalBeanBuildItem.unremovableOf(DebeziumTracerEventDispatcher.class));
}
else {

2
debezium-quarkus-outbox/runtime/src/main/java/io/debezium/outbox/quarkus/internal/AbstractEventDispatcher.java
View File

@ -47,7 +47,7 @@ protected Void persist(Map<String, Object> dataMap) {
private void remove(Map<String, Object> dataMap, Session session) {
// Remove entity if the configuration deems doing so, leaving useful for debugging
if (config.removeAfterInsert) {
if (config.removeAfterInsert()) {
session.delete(OUTBOX_ENTITY_FULLNAME, dataMap);
}
}

6
debezium-quarkus-outbox/runtime/src/main/java/io/debezium/outbox/quarkus/internal/DebeziumOutboxRuntimeConfig.java
View File

@ -7,13 +7,15 @@
import io.quarkus.runtime.annotations.ConfigPhase;
import io.quarkus.runtime.annotations.ConfigRoot;
import io.smallrye.config.ConfigMapping;
/**
* Debezium outbox Quarkus extension runtime configuration properties.
*
* @author Chris Cranford
*/
@ConfigRoot(phase = ConfigPhase.RUN_TIME, name = "debezium-outbox")
public class DebeziumOutboxRuntimeConfig extends DebeziumOutboxCommonRuntimeConfig {
@ConfigMapping(prefix = "quarkus.debezium-outbox")
@ConfigRoot(phase = ConfigPhase.RUN_TIME)
public interface DebeziumOutboxRuntimeConfig extends DebeziumOutboxCommonRuntimeConfig {
}

2
pom.xml
View File

@ -125,7 +125,7 @@
<!-- Quarkus -->
<!-- Version used for Outbox extension development, changes frequently to align with latest Quarkus -->
<quarkus.version.extension>3.12.0</quarkus.version.extension>
<quarkus.version.extension>3.14.0</quarkus.version.extension>
<!-- Version used in Debezium Server, Operator, etc., usually a LTS version -->
<!-- Must be aligned with Apicurio version below -->