Mailing-List: contact jira-help@kafka.apache.org; run by ezmlm
Precedence: bulk
Reply-To: jira@kafka.apache.org
Date: Tue, 27 Jun 2017 22:24:00 +0000 (UTC)
From: "Matthias J. Sax (JIRA)" <jira@apache.org>
To: jira@kafka.apache.org
Message-ID: <JIRA.13082944.1498597423000.119249.1498602240119@Atlassian.JIRA>
In-Reply-To: <JIRA.13082944.1498597423000@Atlassian.JIRA>
References: <JIRA.13082944.1498597423000@Atlassian.JIRA> <JIRA.13082944.1498597423158@jira-lw-us.apache.org>
Subject: [jira] [Updated] (KAFKA-5528) Error while reading topic, offset,
 partition info from process method
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
archived-at: Tue, 27 Jun 2017 22:24:08 -0000


     [ https://issues.apache.org/jira/browse/KAFKA-5528?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Matthias J. Sax updated KAFKA-5528:
-----------------------------------
    Description: 
We are encountering an {{IllegalStateException}} while trying to access {{context.topic()}} from process function. The code is written in Scala and is being launched using sbt (spring isn't involved). Here's the code sketch:

{noformat}
class GenericProcessor[T <: ThriftStruct](serDe: ThriftStructSerDe[T], decrypt: Boolean, config: Config) extends AbstractProcessor[Array[Byte], Array[Byte]] with LazyLogging {
  private var hsmClient: HSMClient = _
  override def init(processorContext: ProcessorContext): Unit = { 
    super.init(processorContext) 
    hsmClient = HSMClient(config).getOrElse(null) 
  }
  override def process(key: Array[Byte], value: Array[Byte]): Unit = { 
    val topic: String = this.context.topic() 
    partition: Int = this.context.partition() 
    val offset: Long = this.context.offset() 
    val timestamp: Long = this.context.timestamp() 
    // business logic 
  }
}
{noformat}

The exception is thrown only for the multi-consumer case (when number of partitions for a topic > 1 and parallelism > 1). 

  was:
We are encountering an IllegalStateException while trying to access context.topic() from process function. The code is written in Scala and is being launched using sbt (spring isn't involved). Here's the code sketch:

class GenericProcessor[T <: ThriftStruct](serDe: ThriftStructSerDe[T], decrypt: Boolean, config: Config) extends AbstractProcessor[Array[Byte], Array[Byte]] with LazyLogging {
  private var hsmClient: HSMClient = _
  override def init(processorContext: ProcessorContext): Unit = { 
    super.init(processorContext) 
    hsmClient = HSMClient(config).getOrElse(null) 
  }
  override def process(key: Array[Byte], value: Array[Byte]): Unit = { 
    val topic: String = this.context.topic() 
    partition: Int = this.context.partition() 
    val offset: Long = this.context.offset() 
    val timestamp: Long = this.context.timestamp() 
    // business logic 
  }
}
The exception is thrown only for the multi-consumer case (when number of partitions for a topic > 1 and parallelism > 1). 


> Error while reading topic, offset, partition info from process method
> ---------------------------------------------------------------------
>
>                 Key: KAFKA-5528
>                 URL: https://issues.apache.org/jira/browse/KAFKA-5528
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams
>    Affects Versions: 0.10.2.0
>            Reporter: Nishkam Ravi
>
> We are encountering an {{IllegalStateException}} while trying to access {{context.topic()}} from process function. The code is written in Scala and is being launched using sbt (spring isn't involved). Here's the code sketch:
> {noformat}
> class GenericProcessor[T <: ThriftStruct](serDe: ThriftStructSerDe[T], decrypt: Boolean, config: Config) extends AbstractProcessor[Array[Byte], Array[Byte]] with LazyLogging {
>   private var hsmClient: HSMClient = _
>   override def init(processorContext: ProcessorContext): Unit = { 
>     super.init(processorContext) 
>     hsmClient = HSMClient(config).getOrElse(null) 
>   }
>   override def process(key: Array[Byte], value: Array[Byte]): Unit = { 
>     val topic: String = this.context.topic() 
>     partition: Int = this.context.partition() 
>     val offset: Long = this.context.offset() 
>     val timestamp: Long = this.context.timestamp() 
>     // business logic 
>   }
> }
> {noformat}
> The exception is thrown only for the multi-consumer case (when number of partitions for a topic > 1 and parallelism > 1). 


--
This message was sent by Atlassian JIRA
(v6.4.14#64029)