Sample2.aplf

 Sample2 n;i;start;cr

 ⍝ Topics created with
 ⍝
 ⍝     kafka-topics.sh \
 ⍝       --bootstrap-server localhost:9092 \
 ⍝       --create --topic "animals" \
 ⍝       --partitions 3
 ⍝
 ⍝     kafka-topics.sh \
 ⍝       --bootstrap-server localhost:9092 \
 ⍝       --create --topic "cars" \
 ⍝       --partitions 3

 ⍝ Call to Init function
 Init 'path/to/dir/housing/kafka/shared/lib'

 consumer←⎕NEW Consumer

 consumer.configure'bootstrap.servers' 'localhost:9092'⍝'grs3550.insight.local:9092'
 consumer.configure'client.id' 'bhcgrs3550'
 consumer.configure'group.id' 'dyalog'

 l←consumer.topic_partition
 consumer.set_topic_partition l'animals'
 consumer.set_topic_partition l'cars'
 consumer.subscribe_topic_partition l
 ⎕DL 5

 producer←⎕NEW Producer
 producer.configure'bootstrap.servers' 'localhost:9092'⍝'grs3550.insight.local:9092'
 producer.configure'client.id' 'bhc'


 :For i :In ⍳n
     producer.produce_record ⎕NEW #.Record('animals'(100↑'Payload',⍕i)('key',⍕3|i))
     :If 0=10|i
         producer.update_outstanding
     :EndIf
 :EndFor

 producer.produce_record ⎕NEW #.Record('cars' 'ferrari' 'sportcars')
 producer.update_outstanding

 start←3⊃⎕AI
 :While 1
     cr←consumer.consume_record
     :If 1=⊃cr
     :AndIf 20000>(3⊃⎕AI)-start
         ⎕DL 0.2
         :Continue
     :ElseIf 0=⊃cr
         (2⊃cr).(Topic Payload Key Partition)
         start←0
     :Else
         cr
         :Leave
     :EndIf
 :EndWhile
 ⎕EX'producer'
 ⎕EX'consumer'