Jak utworzyć partycję na temat Kafki za pomocą Samzy? - scala, apache-kafka, partycjonowanie, apache-samza

Mam kilka zadań Samzy, na których czytam wszystkie pracewiadomości z tematu Kafki i pisanie nowej wiadomości na nowy temat. Aby wysłać nowe wiadomości, używam wbudowanej Samzy w OutgoingMessageEnvelope. Używając kolektora wiadomości, aby wysłać nową wiadomość. Wygląda to mniej więcej tak:

collector.send(new OutgoingMessageEnvelope(SystemStream, newMessage))

Czy istnieje sposób, aby użyć tego do dodania partycji do tematu Kafki? Takie jak partycjonowanie na identyfikatorze użytkownika lub coś takiego.

Albo jeśli jest lepszy sposób, chciałbym to usłyszeć!

Odpowiedzi:

Powinieneś być w stanie wysyłać wiadomości za pomocą klucz partycjonowania,

    public OutgoingMessageEnvelope(SystemStream systemStream,
java.lang.Object partitionKey,
java.lang.Object key,
java.lang.Object message)
Constructs a new OutgoingMessageEnvelope from specified components.
Parameters:
systemStream - Object representing the appropriate stream of which this envelope will be sent on.
partitionKey - A key representing which partition of the systemStream to send this envelope on.
key - A deserialized key to be used for the message.
message - A deserialized message to be sent in this envelope.

Użycie tej metody spowoduje podzielenie danych. Jednak myślę, że jeśli chcesz kontrolować programowo liczbę partycji, powinieneś użyć API kafka do utworzenia / zmiany tematu, jak wspomniano tutaj