I think you got your imports wrong. Given the call to .asJava(), I'm
guessing you've imported Consumer from the scaladsl.
If you use javadsl you'll get akka.stream.javadsl.Source, not
akka.stream.scaladsl.Source in the Tuple2.
committablePartitionedSource() is giving you a Source of tuples, one per
partition, the second element of each tuple is the actual Source of
messages.
If you just want to a simple app, then committableSource() directly
gives you a Source of messages, without the breakdown per partition
which you probably don't need at this stage, but it's up to you.
Examples are in the docs:
http://doc.akka.io/docs/akka-stream-kafka/current/consumer.html
Hope that helps,
Michal
On 01/05/17 23:02, Shannon Ma wrote:
---made some changes, still with mapAsync, in the apply() the input is
Source not Message, how can i get the message from it, i cannot find
anything.
TopicPartition par = new TopicPartition("test", 3);
Source<Tuple2<TopicPartition,
akka.stream.scaladsl.Source<CommittableMessage<SpecificRecord,
SpecificRecord>, NotUsed>>, Control> s
= Consumer.committablePartitionedSource(consumerSettings,
Subscriptions.topics("test")).asJava();
s.mapAsync(3, new
Function<Tuple2<TopicPartition,akka.stream.scaladsl.Source<CommittableMessage<SpecificRecord,SpecificRecord>,NotUsed>>,
CompletionStage<ProducerMessage.Message<SpecificRecord,SpecificRecord,
ConsumerMessage.Committable>>>() {
public CompletionStage<Message<SpecificRecord, SpecificRecord,
Committable>> apply(
Tuple2<TopicPartition,
akka.stream.scaladsl.Source<CommittableMessage<SpecificRecord,
SpecificRecord>, NotUsed>> param)
throws Exception {
System.out.println("thread == " + Thread.currentThread().getName() + "
part is ----------------" + param._1);
System.out.println("thread == " + Thread.currentThread().getName() + "
value is ----------------" + param._2.asJava());
//System.out.println("thread == " + Thread.currentThread().getName() +
"msg----------------" + msg);
return null;
//return new ProducerMessage.Message<SpecificRecord, SpecificRecord,
ConsumerMessage.Committable>(
//new ProducerRecord<SpecificRecord, SpecificRecord>("akkatest",
param._2.record().key(), msg.record().value()), msg.committableOffset());
}
}).runWith(Producer.commitableSink(producerSettings), materializer);
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google
Groups "Akka User List" group.
To unsubscribe from this group and stop receiving emails from it, send
an email to [email protected]
<mailto:[email protected]>.
To post to this group, send email to [email protected]
<mailto:[email protected]>.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
--
Signature
<http://www.openbet.com/> Michal Borowiecki
Senior Software Engineer L4
T: +44 208 742 1600
+44 203 249 8448
E: [email protected]
W: www.openbet.com <http://www.openbet.com/>
OpenBet Ltd
Chiswick Park Building 9
566 Chiswick High Rd
London
W4 5XT
UK
<https://www.openbet.com/email_promo>
This message is confidential and intended only for the addressee. If you
have received this message in error, please immediately notify the
[email protected] <mailto:[email protected]> and delete it
from your system as well as any copies. The content of e-mails as well
as traffic data may be monitored by OpenBet for employment and security
purposes. To protect the environment please do not print this e-mail
unless necessary. OpenBet Ltd. Registered Office: Chiswick Park Building
9, 566 Chiswick High Road, London, W4 5XT, United Kingdom. A company
registered in England and Wales. Registered no. 3134634. VAT no.
GB927523612
--
Read the docs: http://akka.io/docs/
Check the FAQ: http://doc.akka.io/docs/akka/current/additional/faq.html
Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
