Uploaded image for project: 'Debezium'
  1. Debezium
  2. DBZ-4111

projectId not being set when injecting a custom PublisherBuilder

    XMLWordPrintable

Details

    • Hide

      Provide a custom publisher to PubSubChangeConsumer like this:

      @Dependent
      @CustomConsumerBuilder
      public class PubSubPublisher implements PubSubChangeConsumer.PublisherBuilder {
          private static final Logger LOGGER = LoggerFactory.getLogger(PubSubPublisher.class);
          @Override
          public Publisher get(ProjectTopicName topicName) {
              LOGGER.info("topic name " + topicName.toString());
              Publisher publisher;
              BatchingSettings batchingSettings = BatchingSettings.newBuilder()
                      .setRequestByteThreshold(Publisher.getApiMaxRequestBytes()) // 1000
                      .setElementCountThreshold(Publisher.getApiMaxRequestElementCount()) // 10MB
                      .setDelayThreshold(Duration.ofMillis(50))
                      .build();
      
              try {
                  publisher = Publisher.newBuilder(topicName)
                          .setEnableMessageOrdering(true)
                          .setBatchingSettings(batchingSettings)
                          .build();
              } catch (IOException e) {
                  e.printStackTrace();
                  return null;
              }
              return publisher;
          };
      }
      

      Run Debezium and make sure that it's being picked up, you should see this in the logs

      Obtained custom configured PublisherBuilder 'io.quarkus.arc.impl.InstanceImpl@7db0565c'
      

      Once Debezium will try and publish events to Pub/Sub, it will crash with

      2021-10-05 11:01:39,724 INFO  [io.deb.ser.ConnectorLifecycle] (pool-7-thread-1) Connector completed: success = 'false', message = 'Stopping connector after error in the application's handler method: null', error = '{}': java.lang.NullPointerException
      	at com.google.common.base.Preconditions.checkNotNull(Preconditions.java:878)
      	at com.google.pubsub.v1.ProjectTopicName.<init>(ProjectTopicName.java:55)
      	at com.google.pubsub.v1.ProjectTopicName.<init>(ProjectTopicName.java:28)
      	at com.google.pubsub.v1.ProjectTopicName$Builder.build(ProjectTopicName.java:156)
      	at com.google.pubsub.v1.ProjectTopicName.of(ProjectTopicName.java:60)
      	at io.debezium.server.pubsub.PubSubChangeConsumer.lambda$handleBatch$2(PubSubChangeConsumer.java:120)
      	at java.base/java.util.HashMap.computeIfAbsent(Unknown Source)
      	at io.debezium.server.pubsub.PubSubChangeConsumer.handleBatch(PubSubChangeConsumer.java:120)
      	at io.debezium.embedded.ConvertingEngineBuilder.lambda$notifying$2(ConvertingEngineBuilder.java:83)
      	at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:821)
      	at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:188)
      	at io.debezium.server.DebeziumServer.lambda$start$1(DebeziumServer.java:145)
      	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
      	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
      	at java.base/java.lang.Thread.run(Unknown Source)
      
      Show
      Provide a custom publisher to PubSubChangeConsumer like this: @Dependent @CustomConsumerBuilder public class PubSubPublisher implements PubSubChangeConsumer.PublisherBuilder { private static final Logger LOGGER = LoggerFactory.getLogger(PubSubPublisher.class); @Override public Publisher get(ProjectTopicName topicName) { LOGGER.info( "topic name " + topicName.toString()); Publisher publisher; BatchingSettings batchingSettings = BatchingSettings.newBuilder() .setRequestByteThreshold(Publisher.getApiMaxRequestBytes()) // 1000 .setElementCountThreshold(Publisher.getApiMaxRequestElementCount()) // 10MB .setDelayThreshold(Duration.ofMillis(50)) .build(); try { publisher = Publisher.newBuilder(topicName) .setEnableMessageOrdering( true ) .setBatchingSettings(batchingSettings) .build(); } catch (IOException e) { e.printStackTrace(); return null ; } return publisher; }; } Run Debezium and make sure that it's being picked up, you should see this in the logs Obtained custom configured PublisherBuilder 'io.quarkus.arc.impl.InstanceImpl@7db0565c' Once Debezium will try and publish events to Pub/Sub, it will crash with 2021-10-05 11:01:39,724 INFO [io.deb.ser.ConnectorLifecycle] (pool-7-thread-1) Connector completed: success = ' false ' , message = 'Stopping connector after error in the application' s handler method: null ', error = ' {}': java.lang.NullPointerException at com.google.common.base.Preconditions.checkNotNull(Preconditions.java:878) at com.google.pubsub.v1.ProjectTopicName.<init>(ProjectTopicName.java:55) at com.google.pubsub.v1.ProjectTopicName.<init>(ProjectTopicName.java:28) at com.google.pubsub.v1.ProjectTopicName$Builder.build(ProjectTopicName.java:156) at com.google.pubsub.v1.ProjectTopicName.of(ProjectTopicName.java:60) at io.debezium.server.pubsub.PubSubChangeConsumer.lambda$handleBatch$2(PubSubChangeConsumer.java:120) at java.base/java.util.HashMap.computeIfAbsent(Unknown Source) at io.debezium.server.pubsub.PubSubChangeConsumer.handleBatch(PubSubChangeConsumer.java:120) at io.debezium.embedded.ConvertingEngineBuilder.lambda$notifying$2(ConvertingEngineBuilder.java:83) at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:821) at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:188) at io.debezium.server.DebeziumServer.lambda$start$1(DebeziumServer.java:145) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) at java.base/java.lang. Thread .run(Unknown Source)

    Description

      When trying to provide a custom PubSub Publisher to PubSubChangeConsumer via @CustomConsumerBuilder injection point, the projectId property of the change consumer is not being set. Because of this return once a custom publisher is found, the projectId is null since it's being set below, here

      It's actually failing here since it's trying to create a TopicName for a null project.

      One way to fix this is to move the lines of code where the project is being set above the if that checks customPublisherBuilder.isResolvable(). Like this

      @PostConstruct    void connect() {
      final Config config = ConfigProvider.getConfig();        
      projectId = config.getOptionalValue(PROP_PROJECT_ID, String.class).orElse(ServiceOptions.getDefaultProjectId());
      
      ... // if statement
      
      }
      

       

       

       Another way is to app new properties to configure the default Publisher with things like batch size/number of messages inside a batch, but this one is going to be difficult.

      Attachments

        Activity

          People

            Unassigned Unassigned
            plugarut Tudor Plugaru
            Votes:
            0 Vote for this issue
            Watchers:
            3 Start watching this issue

            Dates

              Created:
              Updated:
              Resolved: