Sample4.aplf

 Sample4 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
 ⍝
 ⍝     kafka-topics.sh \
 ⍝       --bootstrap-server localhost:9092 \
 ⍝       --create --topic "plants" \
 ⍝       --partitions 3

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

 config←0 2⍴⍬
 config⍪←'bootstrap.servers' 'localhost:9092'
 config⍪←'client.id' 'bhcgrs3550'
 config⍪←'group.id' 'dyalog'

 topic_list←'animals' 'cars' 'plants'

 consumer←⎕NEW Consumer(config topic_list)
 ⎕DL 5

 config←0 2⍴⍬
 config⍪←'bootstrap.servers' 'localhost:9092'
 config⍪←'client.id' 'bhc'

 producer←⎕NEW Producer config

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

 producer.produce_record ⎕NEW #.Record('cars' 'ferrari' 'sportcars')
 producer.produce_record ⎕NEW #.Record('plants' 'iris' 'flowers' 0)
 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'