If you tested earlier with the same stack, which version did you use?
*Can you enable debug logs to check what’s happening there? *So far the
following warning was received from from log4j which I received from log4j
on Databricks (no errors other than that).
*Can you make sure that there is no issue with firewall or something? *No I
don't think so. Because it's working fine locally and databricks notebook.
*Can you run this pipeline locally against a real Kafka server, not Azure
Event Hub, to make sure that it works fine? *Yes it's working fine with
both Azure EventHub and Kafka
org.springframework.core.convert.support.DefaultConversionService.getSharedInstance()'
at
org.springframework.expression.spel.support.StandardTypeConverter.<init>(StandardTypeConverter.java:46)
at
org.springframework.expression.spel.support.StandardEvaluationContext.getTypeConverter(StandardEvaluationContext.java:197)
at
org.springframework.expression.spel.support.ReflectiveMethodResolver.resolve(ReflectiveMethodResolver.java:115)
at
org.springframework.expression.spel.ast.MethodReference.findAccessorForMethod(MethodReference.java:201)
at
org.springframework.expression.spel.ast.MethodReference.getValueInternal(MethodReference.java:130)
at
org.springframework.expression.spel.ast.MethodReference.access$000(MethodReference.java:52)
at
org.springframework.expression.spel.ast.MethodReference$MethodValueRef.getValue(MethodReference.java:377)
at
org.springframework.expression.spel.ast.CompoundExpression.getValueInternal(CompoundExpression.java:88)
at
org.springframework.expression.spel.ast.SpelNodeImpl.getValue(SpelNodeImpl.java:121)
at
org.springframework.expression.spel.standard.SpelExpression.getValue(SpelExpression.java:262)
at
org.apache.beam.sdk.io.kafka.ConsumerSpEL.evaluateAssign(ConsumerSpEL.java:124)
at
org.apache.beam.sdk.io.kafka.KafkaUnboundedReader.start(KafkaUnboundedReader.java:85)
at
org.apache.beam.runners.spark.io.MicrobatchSource$Reader.startIfNeeded(MicrobatchSource.java:207)
at
org.apache.beam.runners.spark.io.MicrobatchSource$Reader.start(MicrobatchSource.java:227)
at
org.apache.beam.runners.spark.stateful.StateSpecFunctions$1.apply(StateSpecFunctions.java:172)
at
org.apache.beam.runners.spark.stateful.StateSpecFunctions$1.apply(StateSpecFunctions.java:111)
at
org.apache.spark.streaming.StateSpec$.$anonfun$function$1(StateSpec.scala:181)
at
org.apache.spark.streaming.rdd.MapWithStateRDDRecord$.$anonfun$updateRecordWithData$3(MapWithStateRDD.scala:57)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at
org.apache.spark.streaming.rdd.MapWithStateRDDRecord$.updateRecordWithData(MapWithStateRDD.scala:55)
at
org.apache.spark.streaming.rdd.MapWithStateRDD.compute(MapWithStateRDD.scala:159)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
at org.apache.spark.rdd.RDD.$anonfun$getOrCompute$1(RDD.scala:393)
at
org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1486)
at org.apache.spark.storage.BlockManager.org
$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1413)
at
org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1477)
at
org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1296)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:391)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:342)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
at
org.apache.spark.scheduler.ResultTask.$anonfun$runTask$3(ResultTask.scala:75)
at
com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
at
org.apache.spark.scheduler.ResultTask.$anonfun$runTask$1(ResultTask.scala:75)
at
com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:55)
at org.apache.spark.scheduler.Task.doRunTask(Task.scala:153)
at org.apache.spark.scheduler.Task.$anonfun$run$1(Task.scala:122)
at
com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
at org.apache.spark.scheduler.Task.run(Task.scala:93)
at
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$13(Executor.scala:824)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1621)
at
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:827)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at
com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:683)
at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)
On Tue, Feb 1, 2022 at 12:07 PM Alexey Romanenko <aromanenko....@gmail.com>
wrote:
> Well, personally I didn’t test with this version, but it should be fine…
> Can you enable debug logs to check what’s happening there?
> Can you make sure that there is no issue with firewall or something?
> Can you run this pipeline locally against a real Kafka server, not Azure
> Event Hub, to make sure that it works fine?
> Otherwise, it would need to debug remotely the worker process.
>
> On 1 Feb 2022, at 19:18, Utkarsh Parekh <utkarsh.s.par...@gmail.com>
> wrote:
>
> Sorry I sent the last message in a hurry. Here is the Beam java to kafka:
> Is something missing here?
>
> <dependency>
> <groupId>org.apache.beam</groupId>
> <artifactId>beam-sdks-java-io-kafka</artifactId>
> <version>2.35.0</version>
> </dependency>
>
>
> On Tue, Feb 1, 2022 at 9:01 AM Utkarsh Parekh <utkarsh.s.par...@gmail.com>
> wrote:
>
>> Here it is
>>
>> <dependency>
>> <groupId>org.apache.kafka</groupId>
>> <artifactId>kafka-clients</artifactId>
>> <version>2.8.0</version>
>> </dependency>
>>
>>
>> On Tue, Feb 1, 2022 at 8:53 AM Alexey Romanenko <aromanenko....@gmail.com>
>> wrote:
>>
>>> Hmm, this is strange. Which version of Kafka client do you use while
>>> running it with Beam?
>>>
>>> On 1 Feb 2022, at 16:56, Utkarsh Parekh <utkarsh.s.par...@gmail.com>
>>> wrote:
>>>
>>> Hi Alexey,
>>>
>>> First of all, thank you for the response! Yes I did have it in Consumer
>>> configuration and try to increase "session.timeout".
>>>
>>> From consumer side so far I've following settings:
>>>
>>> props.put("sasl.mechanism", SASL_MECHANISM);
>>> props.put("security.protocol", SECURITY_PROTOCOL);
>>> props.put("sasl.jaas.config", saslJaasConfig);
>>> props.put("request.timeout.ms", 60000);
>>> props.put("session.timeout.ms", 60000);
>>> props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
>>> AUTO_OFFSET_RESET_CONFIG);
>>> props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
>>>
>>>
>>> It works fine using following code in Databricks Notebook. The problem
>>> has been occurring when I run it through Apache beam and KafkaIO (Just
>>> providing more context if that may help you to understand problem)
>>>
>>> val df = spark.readStream
>>> .format("kafka")
>>> .option("subscribe", TOPIC)
>>> .option("kafka.bootstrap.servers", BOOTSTRAP_SERVERS)
>>> .option("kafka.sasl.mechanism", "PLAIN")
>>> .option("kafka.security.protocol", "SASL_SSL")
>>> .option("kafka.sasl.jaas.config", EH_SASL)
>>> .option("kafka.request.timeout.ms", "60000")
>>> .option("kafka.session.timeout.ms", "60000")
>>> .option("failOnDataLoss", "false")
>>> //.option("kafka.group.id", "testsink")
>>> .option("startingOffsets", "latest")
>>> .load()
>>>
>>> Utkarsh
>>>
>>> On Tue, Feb 1, 2022 at 6:20 AM Alexey Romanenko <
>>> aromanenko....@gmail.com> wrote:
>>>
>>>> Hi Utkarsh,
>>>>
>>>> Can it be related to this configuration problem?
>>>>
>>>> https://docs.microsoft.com/en-us/azure/event-hubs/apache-kafka-troubleshooting-guide#no-records-received
>>>>
>>>> Did you check timeout settings?
>>>>
>>>> —
>>>> Alexey
>>>>
>>>>
>>>> On 1 Feb 2022, at 02:27, Utkarsh Parekh <utkarsh.s.par...@gmail.com>
>>>> wrote:
>>>>
>>>> Hello,
>>>>
>>>> I'm doing POC with KafkaIO and spark runner on Azure Databricks. I'm
>>>> trying to create a simple streaming app with Apache Beam, where it reads
>>>> data from an Azure event hub and produces messages into another Azure event
>>>> hub.
>>>>
>>>> I'm creating and running spark jobs on Azure Databricks.
>>>>
>>>> The problem is the consumer (uses SparkRunner) is not able to read data
>>>> from Event hub (queue). There is no activity and no errors on the Spark
>>>> cluster.
>>>>
>>>> I would appreciate it if anyone could help to fix this issue.
>>>>
>>>> Thank you
>>>>
>>>> Utkarsh
>>>>
>>>>
>>>>
>>>
>
