問題タブ [spring-cloud-stream]

spring-cloud-stream で、アプリケーションを再起動せずにインスタンス数とインスタンス インデックスを変更する方法はありますか?

また、これらの値を自動的に入力するための推奨される方法はありますか? マイクロサービスの世界では、サービスは常に開始と停止を繰り返しているため、これは非常に難しいように思えます。

Kafka にメッセージを送信し、Spring クラウド ストリームを使用して Kafka から同じメッセージを消費しようとしています。Postman を使用して JSON String { "acctNo" : "32432", "tn" : "3234" } をプロデューサーの残りのコントローラーに送信しています。JSON Parserexception を取得しています:

私のコードは次のようなものです:

プロデューサー アプリケーション yaml

プロデューサー レスト コントローラー

プロデューサーコード

コンシューマ アプリケーション Yaml

消費者コード

ドメインクラス

これは何が原因ですか？助けてください。

私のアプリケーションは、あるマシンで実行されている Kafka サーバーからのメッセージを消費し、別のインスタンスで実行されている別のリモート Kafka に転送します。アプリを Cloud Foundry にデプロイし、最初の Kafka サーバーにメッセージを送信すると、アプリケーションは期待どおりに動作します。メッセージは消費され、リモートの Kafka に転送されます。

ただし、その後、Cloud Foundry で以下の例外の無限ループが発生します (ローカル マシンでもより遅いペースで):

スタックトレース：

私のアプリケーションyamlファイルはこのようなものです

アプリケーション YML :

以下の構成を含めるとエラーが消えることを確認しましたが、メッセージが消費されて送信される無限ループが発生しました。

スニペット:

この無限ループを止めるにはどうすればいいですか?

こんにちはマリウス、SOS コールに応答していただきありがとうございます。上記の問題を強化しました。フローは、a1p(topic:test) から消費し、メッセージが有効な場合は a3p(topic:test) に転送し、そうでない場合はエラー メッセージを a1p(topic:errorMsgQueue) に送信します。以下のアプリケーションがあります。yml ファイル

春: クラウド: ストリーム: バインディング: errorMsgQueue: バインダー: kafka1 宛先: errorMsgQueue contentType: アプリケーション/json 入力: バインダー: kafka2 コンテンツ タイプ: application/x-java-object;type=com.comcast.activation.message.vo. ActivationDataInfo 宛先: テスト グループ: prac
activationMsgQueue: バインダー: kafka3 宛先: テスト contentType: アプリケーション/json バインダー: kafka1: タイプ: kafka 環境: 春: クラウド: ストリーム: kafka: バインダー: ブローカー: caapmsg-as-a1p.sys.comcast.net zk-nodes: caapmsg -as-a1p.sys.comcast.net kafka2: タイプ: kafka 環境: 春: クラウド: ストリーム: kafka: バインダー: ブローカー: caapmsg-as-a3p.sys.comcast.net zk-nodes: caapmsg-as-a3p. sys.comcast.net kafka3: タイプ: カフカ 環境: 春: クラウド:ストリーム: kafka: バインダー: ブローカー: caapmsg-as-a1p.sys.comcast.net zk ノード: caapmsg-as-a1p.sys.comcast.net デフォルト バインダー: kafka2

私はまだ無限ループを取得しています。私は何を間違っていますか？

実装したいシナリオは、 Kafka からのメッセージを消費して処理することです。何らかの条件が失敗した場合、メッセージを確認したくありません。これについては、Spring Cloud Stream リファレンス ドキュメントで見つけました。

autoCommitOffset メッセージが処理されたときにオフセットを自動コミットするかどうか。false に設定すると、確認応答ヘッダーがメッセージ ヘッダーで使用可能になり、遅延確認応答が行われます。

デフォルト: 真。

私の質問は、autoCommitOffset を false に設定した後、メッセージを確認するにはどうすればよいですか? コード例は非常に高く評価されます。

I have 3 projects: A hystrix dashboard, a turbine server (using AMQP) and an API

When I start in development env, I set up 2 instances of the API (using port 8080 and 8081). To test the turbine aggregation, I make calls and in the dashboard, I can see Hosts: 2.
Although when I use Docker, even when the load balancer hits the 2 server, I only see one Host on the hystrix dashboard.

My assumptions:
1- as both containers start on the same port (8080), Turbine sees them as one
2- as I also dockerize RabbitMQ, this may be causing problems

here is my docker-compose.yml file

my persona_api config file

I'm afraid that if I deploy it to production (on Rancher or Docker cloud), I'll see the same problem.

here is a GIF of what is happening when I set up two APIs load balanced

以下に示すように、1 つのパーティションと 1 つのレプリケーション ファクターのみを持つトピック car があります。

Topic:cars PartitionCount:1 ReplicationFactor:1 Configs: Topic: cars Partition: 0 Leader: 0 Replicas: 0 Isr: 0

ローカルホストの 9092 で 1 つ、9093 で 1 つの 2 つのブローカーを実行しています。

私の Java アプリケーションはトピック cars にメッセージを送信します。非常に奇妙な動作が見られます。

1. 私の Java アプリケーションは、メッセージ #1、メッセージ #3 などの代替メッセージしか受信しません。

この問題をデバッグするために、コンソール コンシューマーを開始しました。これで、Java アプリケーションが取得できなかったメッセージのみが表示されました。たとえば、メッセージ #2​​、メッセージ #4 など

メッセージがトピックに正しく投稿されていることは明らかですが、Java アプリケーションが代替メッセージしか受信しない原因は何ですか?

そして最後に、なぜコンシューマ コンソールは上記の動作しか示さないのですか?