sasakitoa / embulk-input-kafka Goto Github PK

Helo Again @sasakitoa

I was using the plugin with the next config.yml to move some data from Kafka to a posgreSQL instance:

in:
  type: kafka
  broker_list: my_broker_cluster:9093
  topics:
    - historicPurchases
  key_deserializer: org.apache.kafka.common.serialization.StringDeserializer
  columns:
    - key
    - value
  load_from_beginning: true
  extra_kafka_options:
    auto.offset.reset: earliest
    enable.auto.commit: false
    security.protocol: SSL
    ssl.truststore.location: /Users/jsanz/fluentdData/kafka.client.truststore.jks
    ssl.truststore.password: myPasword
out:
  type: postgresql
  host: localhost
  user: postgres
  password: "myPasword"
  database: test
  table: purchases
  options: {loglevel: 2}
  mode: insert

On the log I was seeing some messages being processed but my table purchases was empty. Using another config.yml with a cvs out I could see mi data. So I finally applied the next patch to the plugin:

--- a/src/main/java/org/embulk/input/kafka/KafkaInputPlugin.java
+++ b/src/main/java/org/embulk/input/kafka/KafkaInputPlugin.java
@@ -116,6 +116,7 @@ public class KafkaInputPlugin implements InputPlugin {
                 showReadRecords *= 2;
             }
         }
+        builder.finish();
         builder.close();
         logger.info(String.format("Finishing task-%d.Total %d record(s) read in this task", taskIndex, readRecords));
         consumer.close();

And that made the difference. Apparently when you close ,finish() is not called. And as on my output I was using insertmode, that mode only executes the commit at the end of the transaction. Seems the commit is executed on finish builder. That line seems to fixed the problem.

The default kafka client this plugin is using supports SSL security transport layer. Here explain how. Could those parameters be exposed on the plugin configuration?

load_from_beginning option will fail when consumer group information is not exists in Kafka.

Error mesage is below.

java.lang.IllegalStateException: No current assignment for partition topic1-0
at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedState(SubscriptionState.java:264)
at org.apache.kafka.clients.consumer.internals.SubscriptionState.needOffsetReset(SubscriptionState.java:336)
at org.apache.kafka.clients.consumer.KafkaConsumer.seekToBeginning(KafkaConsumer.java:1218)
at org.embulk.input.kafka.KafkaInputPlugin.setOffsetPosition(KafkaInputPlugin.java:138)
at org.embulk.input.kafka.KafkaInputPlugin.run(KafkaInputPlugin.java:101)
at org.embulk.exec.PreviewExecutor$2$1.run(PreviewExecutor.java:141)
at org.embulk.spi.util.Filters$RecursiveControl.transaction(Filters.java:96)
at org.embulk.spi.util.Filters.transaction(Filters.java:49)
at org.embulk.exec.PreviewExecutor$2.run(PreviewExecutor.java:130)
at org.embulk.input.kafka.KafkaInputPlugin.resume(KafkaInputPlugin.java:80)
at org.embulk.input.kafka.KafkaInputPlugin.transaction(KafkaInputPlugin.java:75)
at org.embulk.exec.PreviewExecutor.doPreview(PreviewExecutor.java:127)
at org.embulk.exec.PreviewExecutor.doPreview(PreviewExecutor.java:114)
at org.embulk.exec.PreviewExecutor.access$000(PreviewExecutor.java:31)
at org.embulk.exec.PreviewExecutor$1.run(PreviewExecutor.java:83)
at org.embulk.exec.PreviewExecutor$1.run(PreviewExecutor.java:79)
at org.embulk.spi.Exec.doWith(Exec.java:25)
at org.embulk.exec.PreviewExecutor.preview(PreviewExecutor.java:79)
at org.embulk.EmbulkEmbed.preview(EmbulkEmbed.java:169)
at org.embulk.EmbulkRunner.previewInternal(EmbulkRunner.java:247)
at org.embulk.EmbulkRunner.preview(EmbulkRunner.java:119)
at org.embulk.cli.EmbulkRun.runSubcommand(EmbulkRun.java:465)
at org.embulk.cli.EmbulkRun.run(EmbulkRun.java:100)

In this case, I specified topic1 for topic.

In document, seek option is not required, but it's needed.

Error mesasge is below

org.embulk.config.ConfigException: com.fasterxml.jackson.databind.JsonMappingException: Field 'seek' is required but not set
at [Source: N/A; line: -1, column: -1]
at org.embulk.config.ModelManager.readObjectWithConfigSerDe(ModelManager.java:75)
at org.embulk.config.DataSourceImpl.loadConfig(DataSourceImpl.java:220)
at org.embulk.input.kafka.KafkaInputPlugin.transaction(KafkaInputPlugin.java:70)
at org.embulk.exec.PreviewExecutor.doPreview(PreviewExecutor.java:127)
at org.embulk.exec.PreviewExecutor.doPreview(PreviewExecutor.java:114)
at org.embulk.exec.PreviewExecutor.access$000(PreviewExecutor.java:31)
at org.embulk.exec.PreviewExecutor$1.run(PreviewExecutor.java:83)
at org.embulk.exec.PreviewExecutor$1.run(PreviewExecutor.java:79)
at org.embulk.spi.Exec.doWith(Exec.java:25)
at org.embulk.exec.PreviewExecutor.preview(PreviewExecutor.java:79)
at org.embulk.EmbulkEmbed.preview(EmbulkEmbed.java:169)
at org.embulk.EmbulkRunner.previewInternal(EmbulkRunner.java:247)
at org.embulk.EmbulkRunner.preview(EmbulkRunner.java:119)
at org.embulk.cli.EmbulkRun.runSubcommand(EmbulkRun.java:465)
at org.embulk.cli.EmbulkRun.run(EmbulkRun.java:100)
at org.embulk.cli.Main.main(Main.java:28)
Caused by: com.fasterxml.jackson.databind.JsonMappingException: Field 'seek' is required but not set
at [Source: N/A; line: -1, column: -1]
at org.embulk.config.TaskSerDe$TaskDeserializer.deserialize(TaskSerDe.java:181)
at com.fasterxml.jackson.databind.ObjectMapper._readValue(ObjectMapper.java:3708)
at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:2005)
at org.embulk.config.ModelManager.readObjectWithConfigSerDe(ModelManager.java:72)
... 15 more

Hi, could you please write me how to install this plugin to Embulk? When I run embulk gem install embulk-input-kafka it doesn't work.

I suspect this is an intended behaviour. But a warning might help.

My use case is that I had a "double" field that could be missing (latitude,longitude). And this was dropping the records. I set it to string to workaround this for testing.