# Signaling activity from workflow

**URL:** <https://community.temporal.io/t/signaling-activity-from-workflow/1079>\
**Category:** Community Support\
**Created:** [November 25, 2020, 8:49am UTC](https://community.temporal.io/t/signaling-activity-from-workflow/1079 "2020-11-25T08:49:39Z")\
**Posts on this page:** 10\
**Page:** 1

<div class="post-metadata">

**Author:** ![Severyn\_Lisovsky](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/severyn_lisovsky/32/421_2.png) [@Severyn\_Lisovsky](https://community.temporal.io/u/Severyn_Lisovsky)\
**Post date:** [November 25, 2020, 8:49am UTC](https://community.temporal.io/t/signaling-activity-from-workflow/1079/1 "2020-11-25T08:49:39Z")

</div>

My use case - I have two sets of activities - one for produces events in external system (let’s say Kafka) and another for consuming these events. Both activities are working simultaneously. Is there a mechanism for notifying consumer activity that producer activity ended and **once you process all remaining events** from external system you can exit, and not wait for other events to be added?

In normal case (when workers are not failing) canceling consumer activities context will do the job. But is there more reliable way of doing that?

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [November 25, 2020, 5:49pm UTC](https://community.temporal.io/t/signaling-activity-from-workflow/1079/2 "2020-11-25T17:49:29Z")

</div>

I’m not sure I understand your design. What do you mean by activities that consume Kafka events? We recommend having an independent Kafka consumer that consumes events and sends them as signals to the workflows that care about them.

---

<div class="post-metadata">

**Author:** ![Severyn\_Lisovsky](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/severyn_lisovsky/32/421_2.png) [@Severyn\_Lisovsky](https://community.temporal.io/u/Severyn_Lisovsky)\
**Post date:** [November 25, 2020, 6:50pm UTC](https://community.temporal.io/t/signaling-activity-from-workflow/1079/3 "2020-11-25T18:50:27Z")

</div>

In my use case Kafka is just a third party system not related to Temporal (could be anything else)

in pseudocode my workflow looks like this

```auto
consumer := workflow.ExecuteActivity(ctx, ConsumerActivity)

ch := workflow.Channel()
for i:=0; i< numPublishers; i++ {
   producer := workflow.ExecuteActivity(ctx, ProducerActivity)
   workflow.Go(ctx, func(){
      var numRows int 
      producer.Get(ctx, &numRows)
      ch.Send(ctx, numRows)
   })
}

var rowsProduced int
for i:=0; i< numPublishers; i++ {
  var res int
  ch.Receive(ctx, &v)
  rowsProduced += res
}

// signal consumer somehow that all producers ended working
// HOW TO DO THAT?

var rowsProcessed int
consumer.Get(ctx, &rowsProcessed)

return rowsProduced, rowsProcessed, nil 

```

where the consumer activity looks like

```auto
var numRows int
for {
item, err := externalQueue.Pop()
if err == io.EOF{
     if signalFromWorkflow received {
          return numRows
     }
     time.Sleep(time.Second)
     continue
}
doProcessing(item)
numRows++
}

```

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [November 25, 2020, 7:29pm UTC](https://community.temporal.io/t/signaling-activity-from-workflow/1079/4 "2020-11-25T19:29:46Z")

</div>

Are you creating a Kafka stream per workflow instance? Then your design makes sense.

But if you have a single Kafka stream for all workflow instances then you don’t need to run the consumer in an activity. It can be just always running consumer that signals workflow upon receiving a message.

---

<div class="post-metadata">

**Author:** ![Severyn\_Lisovsky](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/severyn_lisovsky/32/421_2.png) [@Severyn\_Lisovsky](https://community.temporal.io/u/Severyn_Lisovsky)\
**Post date:** [November 25, 2020, 9:32pm UTC](https://community.temporal.io/t/signaling-activity-from-workflow/1079/5 "2020-11-25T21:32:34Z")

</div>

> [@maxim](#):
>
> Are you creating a Kafka stream per workflow instance? Then your design makes sense.

yes, exactly - single stream per workflow

the only difficulty I have is to signal consumer activity that it can no longer wait after consuming all the messages from stream.

There is of course a workaround - to add activity after publishing that will write dummy EOF message in the end of the stream. But just wanted to know if there is any better option?

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [November 25, 2020, 9:47pm UTC](https://community.temporal.io/t/signaling-activity-from-workflow/1079/6 "2020-11-25T21:47:12Z")

</div>

Interesting. Have you considered not creating such stream and just signal workflow directly? Kafka stream per workflow instance sounds like a very heavyweight solution.

Independently, if you decide to stick to your current design just cancel the second activity. It is done by cancelling its context. Activity must heartbeat and have reasonably small heartbeat timeout to be cancellable. [Here is](https://github.com/temporalio/samples-go/tree/master/cancelactivity) the relevant sample.

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [November 25, 2020, 9:49pm UTC](https://community.temporal.io/t/signaling-activity-from-workflow/1079/7 "2020-11-25T21:49:57Z")

</div>

Sorry, I looked at the sample and while it demonstrates how to implement an activity to be cancellable it doesn’t demonstrate how to cancel it explicitly. Use `workflow.WithCancel` to create a cancellable context.

---

<div class="post-metadata">

**Author:** ![Severyn\_Lisovsky](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/severyn_lisovsky/32/421_2.png) [@Severyn\_Lisovsky](https://community.temporal.io/u/Severyn_Lisovsky)\
**Post date:** [November 25, 2020, 10:06pm UTC](https://community.temporal.io/t/signaling-activity-from-workflow/1079/8 "2020-11-25T22:06:39Z")

</div>

workflow.WithCancel was my initial thought, but it wouldn’t work in case when all the producers finished their job and consumer context was cancelled, but not all the messages from stream are consumed yet. Correct me if I’m wrong - if the worker will crash in this moment, consuming activity will not be restarted as the context was already cancelled.

---

<div class="post-metadata">

**Author:** ![Severyn\_Lisovsky](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/severyn_lisovsky/32/421_2.png) [@Severyn\_Lisovsky](https://community.temporal.io/u/Severyn_Lisovsky)\
**Post date:** [November 25, 2020, 10:14pm UTC](https://community.temporal.io/t/signaling-activity-from-workflow/1079/9 "2020-11-25T22:14:53Z")

</div>

Huge thanks @maxim , for answering me. I find Temporal extremely nice tool, which removed a lot of burden from my service! I will for sure recommend it to my colleagues.

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [November 25, 2020, 11:35pm UTC](https://community.temporal.io/t/signaling-activity-from-workflow/1079/10 "2020-11-25T23:35:51Z")

</div>

You are welcome!

> workflow.WithCancel was my initial thought, but it wouldn’t work in case when all the producers finished their job and consumer context was cancelled, but not all the messages from stream are consumed yet. Correct me if I’m wrong - if the worker will crash in this moment, consuming activity will not be restarted as the context was already cancelled.

Then you have to indicate through the stream that all messages are delivered. It looks like the workflow code doesn’t know when to stop the consuming activity. IMHO you would be much better just sending signals to the workflow without Kafka.