Destination Redis: "JedisMovedDataException" when running Redshift to Redis connection

israel.mendes · July 1, 2022, 6:04pm

Is this your first time deploying Airbyte?: No
OS Version / Instance: Ubuntu / AWS EC2
Memory / Disk: m5.2xlarge - 8 CPU, 32 GB Memory
Deployment: Docker
Airbyte Version: 0.39.14-alpha
Source name/version: airbyte/source-redshift 0.3.10
Destination name/version: airbyte/destination-redis 0.1.2
Step:
1. Configured the source (Redshift) and the destination (Redis). All running fine so far.
2. Create the connection and sync it.
3. Error when running the connection:
Description: I tried to read the documentation here, but it appears that it is incomplete:

image766×113 3.78 KB
Logs from the failed connection:

Logs


2022-07-01 16:58:09 e[44msourcee[0m > 2022-07-01 16:58:09 e[32mINFOe[m i.a.i.s.r.AbstractDbSource(lambda$read$2):126 - Closed database connection pool.
2022-07-01 16:58:09 e[44msourcee[0m > 2022-07-01 16:58:09 e[32mINFOe[m i.a.i.b.IntegrationRunner(runInternal):169 - Completed integration: io.airbyte.integrations.source.redshift.RedshiftSource
2022-07-01 16:58:09 e[44msourcee[0m > 2022-07-01 16:58:09 e[32mINFOe[m i.a.i.s.r.RedshiftSource(main):126 - completed source: class io.airbyte.integrations.source.redshift.RedshiftSource
2022-07-01 16:58:09 e[43mdestinatione[0m > 2022-07-01 16:58:09 e[1;31mERRORe[m i.a.i.b.FailureTrackingAirbyteMessageConsumer(accept):52 - Exception while accepting message
2022-07-01 16:58:09 e[43mdestinatione[0m > redis.clients.jedis.exceptions.JedisMovedDataException: MOVED 480 10.4.12.55:6379
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at redis.clients.jedis.Protocol.processError(Protocol.java:121) ~[jedis-3.7.0.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at redis.clients.jedis.Protocol.process(Protocol.java:173) ~[jedis-3.7.0.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at redis.clients.jedis.Protocol.read(Protocol.java:227) ~[jedis-3.7.0.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at redis.clients.jedis.Connection.readProtocolWithCheckingBroken(Connection.java:352) ~[jedis-3.7.0.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at redis.clients.jedis.Connection.getIntegerReply(Connection.java:294) ~[jedis-3.7.0.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at redis.clients.jedis.Jedis.incr(Jedis.java:772) ~[jedis-3.7.0.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.destination.redis.RedisHCache.insert(RedisHCache.java:30) ~[io.airbyte.airbyte-integrations.connectors-destination-redis-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.destination.redis.RedisMessageConsumer.acceptTracked(RedisMessageConsumer.java:66) ~[io.airbyte.airbyte-integrations.connectors-destination-redis-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer.accept(FailureTrackingAirbyteMessageConsumer.java:50) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.consumeWriteStream(IntegrationRunner.java:194) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.lambda$runConsumer$4(IntegrationRunner.java:203) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.watchForOrphanThreads(IntegrationRunner.java:232) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.runConsumer(IntegrationRunner.java:202) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.lambda$runInternal$1(IntegrationRunner.java:165) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.sentry.AirbyteSentry.executeWithTracing(AirbyteSentry.java:54) [io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.sentry.AirbyteSentry.executeWithTracing(AirbyteSentry.java:38) [io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.java:165) [io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.run(IntegrationRunner.java:107) [io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.destination.redis.RedisDestination.main(RedisDestination.java:24) [io.airbyte.airbyte-integrations.connectors-destination-redis-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 2022-07-01 16:58:09 e[33mWARNe[m i.a.i.b.FailureTrackingAirbyteMessageConsumer(close):63 - Airbyte message consumer: failed.
2022-07-01 16:58:09 e[43mdestinatione[0m > 2022-07-01 16:58:09 e[1;31mERRORe[m i.a.i.b.AirbyteExceptionHandler(uncaughtException):26 - Something went wrong in the connector. See the logs for more details.
2022-07-01 16:58:09 e[43mdestinatione[0m > redis.clients.jedis.exceptions.JedisMovedDataException: MOVED 480 10.4.12.55:6379
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at redis.clients.jedis.Protocol.processError(Protocol.java:121) ~[jedis-3.7.0.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at redis.clients.jedis.Protocol.process(Protocol.java:173) ~[jedis-3.7.0.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at redis.clients.jedis.Protocol.read(Protocol.java:227) ~[jedis-3.7.0.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at redis.clients.jedis.Connection.readProtocolWithCheckingBroken(Connection.java:352) ~[jedis-3.7.0.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at redis.clients.jedis.Connection.getIntegerReply(Connection.java:294) ~[jedis-3.7.0.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at redis.clients.jedis.Jedis.incr(Jedis.java:772) ~[jedis-3.7.0.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.destination.redis.RedisHCache.insert(RedisHCache.java:30) ~[io.airbyte.airbyte-integrations.connectors-destination-redis-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.destination.redis.RedisMessageConsumer.acceptTracked(RedisMessageConsumer.java:66) ~[io.airbyte.airbyte-integrations.connectors-destination-redis-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer.accept(FailureTrackingAirbyteMessageConsumer.java:50) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.consumeWriteStream(IntegrationRunner.java:194) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.lambda$runConsumer$4(IntegrationRunner.java:203) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.watchForOrphanThreads(IntegrationRunner.java:232) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.runConsumer(IntegrationRunner.java:202) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.lambda$runInternal$1(IntegrationRunner.java:165) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.sentry.AirbyteSentry.executeWithTracing(AirbyteSentry.java:54) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.sentry.AirbyteSentry.executeWithTracing(AirbyteSentry.java:38) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.java:165) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.base.IntegrationRunner.run(IntegrationRunner.java:107) ~[io.airbyte.airbyte-integrations.bases-base-java-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[43mdestinatione[0m > 	at io.airbyte.integrations.destination.redis.RedisDestination.main(RedisDestination.java:24) ~[io.airbyte.airbyte-integrations.connectors-destination-redis-0.39.20-alpha.jar:?]
2022-07-01 16:58:09 e[32mINFOe[m i.a.w.g.DefaultReplicationWorker(lambda$getDestinationOutputRunnable$7):405 - State in DefaultReplicationWorker from destination: io.airbyte.protocol.models.AirbyteMessage@10e030b0[type=TRACE,log=<null>,spec=<null>,connectionStatus=<null>,catalog=<null>,record=<null>,state=<null>,trace=io.airbyte.protocol.models.AirbyteTraceMessage@1e27ccdc[type=ERROR,emittedAt=1.656694689164E12,error=io.airbyte.protocol.models.AirbyteErrorTraceMessage@1a185bc0[message=Something went wrong in the connector. See the logs for more details.,internalMessage=redis.clients.jedis.exceptions.JedisMovedDataException: MOVED 480 10.4.12.55:6379,stackTrace=redis.clients.jedis.exceptions.JedisMovedDataException: MOVED 480 10.4.12.55:6379
	at redis.clients.jedis.Protocol.processError(Protocol.java:121)
	at redis.clients.jedis.Protocol.process(Protocol.java:173)
	at redis.clients.jedis.Protocol.read(Protocol.java:227)
	at redis.clients.jedis.Connection.readProtocolWithCheckingBroken(Connection.java:352)
	at redis.clients.jedis.Connection.getIntegerReply(Connection.java:294)
	at redis.clients.jedis.Jedis.incr(Jedis.java:772)
	at io.airbyte.integrations.destination.redis.RedisHCache.insert(RedisHCache.java:30)
	at io.airbyte.integrations.destination.redis.RedisMessageConsumer.acceptTracked(RedisMessageConsumer.java:66)
	at io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer.accept(FailureTrackingAirbyteMessageConsumer.java:50)
	at io.airbyte.integrations.base.IntegrationRunner.consumeWriteStream(IntegrationRunner.java:194)
	at io.airbyte.integrations.base.IntegrationRunner.lambda$runConsumer$4(IntegrationRunner.java:203)
	at io.airbyte.integrations.base.IntegrationRunner.watchForOrphanThreads(IntegrationRunner.java:232)
	at io.airbyte.integrations.base.IntegrationRunner.runConsumer(IntegrationRunner.java:202)
	at io.airbyte.integrations.base.IntegrationRunner.lambda$runInternal$1(IntegrationRunner.java:165)
	at io.airbyte.integrations.base.sentry.AirbyteSentry.executeWithTracing(AirbyteSentry.java:54)
	at io.airbyte.integrations.base.sentry.AirbyteSentry.executeWithTracing(AirbyteSentry.java:38)
	at io.airbyte.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.java:165)
	at io.airbyte.integrations.base.IntegrationRunner.run(IntegrationRunner.java:107)
	at io.airbyte.integrations.destination.redis.RedisDestination.main(RedisDestination.java:24)
,failureType=system_error,additionalProperties={}],additionalProperties={}],additionalProperties={}]
2022-07-01 16:58:09 e[32mINFOe[m i.a.w.g.DefaultReplicationWorker(lambda$getReplicationRunnable$6):347 - Source has no more messages, closing connection.
2022-07-01 16:58:09 e[32mINFOe[m i.a.w.g.DefaultReplicationWorker(lambda$getReplicationRunnable$6):355 - Total records read: 23 (26 KB)
2022-07-01 16:58:09 e[1;31mERRORe[m i.a.w.g.DefaultReplicationWorker(run):180 - Sync worker failed.
java.util.concurrent.ExecutionException: io.airbyte.workers.general.DefaultReplicationWorker$DestinationException: Destination process exited with non-zero exit code 1
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396) ~[?:?]
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073) ~[?:?]
	at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:173) ~[io.airbyte-airbyte-workers-0.39.14-alpha.jar:?]
	at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:65) ~[io.airbyte-airbyte-workers-0.39.14-alpha.jar:?]
	at io.airbyte.workers.temporal.TemporalAttemptExecution.lambda$getWorkerThread$2(TemporalAttemptExecution.java:158) ~[io.airbyte-airbyte-workers-0.39.14-alpha.jar:?]
	at java.lang.Thread.run(Thread.java:833) [?:?]
	Suppressed: java.io.IOException: Stream closed
		at java.lang.ProcessBuilder$NullOutputStream.write(ProcessBuilder.java:445) ~[?:?]
		at java.io.OutputStream.write(OutputStream.java:162) ~[?:?]
		at java.io.BufferedOutputStream.write(BufferedOutputStream.java:123) ~[?:?]
		at sun.nio.cs.StreamEncoder.writeBytes(StreamEncoder.java:234) ~[?:?]
		at sun.nio.cs.StreamEncoder.implWrite(StreamEncoder.java:304) ~[?:?]
		at sun.nio.cs.StreamEncoder.implWrite(StreamEncoder.java:282) ~[?:?]
		at sun.nio.cs.StreamEncoder.write(StreamEncoder.java:132) ~[?:?]
		at java.io.OutputStreamWriter.write(OutputStreamWriter.java:205) ~[?:?]
		at java.io.BufferedWriter.flushBuffer(BufferedWriter.java:120) ~[?:?]
		at java.io.BufferedWriter.flush(BufferedWriter.java:256) ~[?:?]
		at io.airbyte.workers.internal.DefaultAirbyteDestination.notifyEndOfInput(DefaultAirbyteDestination.java:98) ~[io.airbyte-airbyte-workers-0.39.14-alpha.jar:?]
		at io.airbyte.workers.internal.DefaultAirbyteDestination.close(DefaultAirbyteDestination.java:111) ~[io.airbyte-airbyte-workers-0.39.14-alpha.jar:?]
		at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:137) ~[io.airbyte-airbyte-workers-0.39.14-alpha.jar:?]
		at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:65) ~[io.airbyte-airbyte-workers-0.39.14-alpha.jar:?]
		at io.airbyte.workers.temporal.TemporalAttemptExecution.lambda$getWorkerThread$2(TemporalAttemptExecution.java:158) ~[io.airbyte-airbyte-workers-0.39.14-alpha.jar:?]
		at java.lang.Thread.run(Thread.java:833) [?:?]
Caused by: io.airbyte.workers.general.DefaultReplicationWorker$DestinationException: Destination process exited with non-zero exit code 1
	at io.airbyte.workers.general.DefaultReplicationWorker.lambda$getDestinationOutputRunnable$7(DefaultReplicationWorker.java:410) ~[io.airbyte-airbyte-workers-0.39.14-alpha.jar:?]
	at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
	... 1 more
2022-07-01 16:58:09 e[32mINFOe[m i.a.w.g.DefaultReplicationWorker(run):239 - sync summary: io.airbyte.config.ReplicationAttemptSummary@d06f2ef[status=failed,recordsSynced=23,bytesSynced=26916,startTime=1656694685805,endTime=1656694689334,totalStats=io.airbyte.config.SyncStats@4ee2ff79[recordsEmitted=23,bytesEmitted=26916,stateMessagesEmitted=0,recordsCommitted=0],streamStats=[io.airbyte.config.StreamSyncStats@451bdb8b[streamName=active_order_vendors,stats=io.airbyte.config.SyncStats@7ef5abcc[recordsEmitted=23,bytesEmitted=26916,stateMessagesEmitted=<null>,recordsCommitted=<null>]]]]
2022-07-01 16:58:09 e[32mINFOe[m i.a.w.g.DefaultReplicationWorker(run):268 - Source did not output any state messages
2022-07-01 16:58:09 e[33mWARNe[m i.a.w.g.DefaultReplicationWorker(run):279 - State capture: No state retained.
2022-07-01 16:58:09 e[32mINFOe[m i.a.w.t.TemporalAttemptExecution(get):134 - Stopping cancellation check scheduling...
2022-07-01 16:58:09 e[32mINFOe[m i.a.w.t.s.ReplicationActivityImpl(lambda$replicate$3):157 - sync summary: io.airbyte.config.StandardSyncOutput@2278878d[standardSyncSummary=io.airbyte.config.StandardSyncSummary@677a0da6[status=failed,recordsSynced=23,bytesSynced=26916,startTime=1656694685805,endTime=1656694689334,totalStats=io.airbyte.config.SyncStats@4ee2ff79[recordsEmitted=23,bytesEmitted=26916,stateMessagesEmitted=0,recordsCommitted=0],streamStats=[io.airbyte.config.StreamSyncStats@451bdb8b[streamName=active_order_vendors,stats=io.airbyte.config.SyncStats@7ef5abcc[recordsEmitted=23,bytesEmitted=26916,stateMessagesEmitted=<null>,recordsCommitted=<null>]]]],normalizationSummary=<null>,state=<null>,outputCatalog=io.airbyte.protocol.models.ConfiguredAirbyteCatalog@4832e2b0[streams=[io.airbyte.protocol.models.ConfiguredAirbyteStream@56576fc4[stream=io.airbyte.protocol.models.AirbyteStream@30f5de6c[name=active_order_vendors,jsonSchema={"type":"object","properties":{"id":{"type":"string"},"city":{"type":"string"},"state":{"type":"string"},"is_tpv":{"type":"boolean"},"country":{"type":"string"},"user_id":{"type":"string"},"zipcode":{"type":"string"},"currency":{"type":"string"},"order_id":{"type":"string"},"vendor_id":{"type":"string"},"created_at":{"type":"string"},"charged_tax":{"type":"number"},"location_id":{"type":"string"},"vendor_name":{"type":"string"},"fulfilled_by":{"type":"string"},"charged_total":{"type":"number"},"current_status":{"type":"string"},"fulfillment_tax":{"type":"number"},"organization_id":{"type":"string"},"charged_discount":{"type":"number"},"charged_shipping":{"type":"number"},"charged_subtotal":{"type":"number"},"fulfillment_type":{"type":"number"},"current_status_at":{"type":"string"},"fulfilled_by_type":{"type":"string"},"fulfillment_total":{"type":"number"},"linked_account_id":{"type":"string"},"organization_name":{"type":"string"},"charged_shipping_tax":{"type":"number"},"fulfillment_discount":{"type":"number"},"fulfillment_shipping":{"type":"number"},"fulfillment_subtotal":{"type":"number"},"vendor_owns_liability":{"type":"boolean"},"avalara_transaction_id":{"type":"string"},"is_processed_by_avalara":{"type":"boolean"},"linked_account_username":{"type":"string"},"dynamic_sourcing_revenue":{"type":"number"},"num_fulfillment_requests":{"type":"number"},"payment_rules_validation":{"type":"string"},"fulfillment_email_sent_to":{"type":"string"},"pricing_set_automatically":{"type":"boolean"},"early_pay_discount_revenue":{"type":"number"},"requested_for_fulfillment_by":{"type":"string"},"fulfillment_exception_present":{"type":"boolean"},"linked_account_owned_by_customer":{"type":"boolean"},"last_requested_for_fulfillment_at":{"type":"string"},"order_eligible_for_fulfillment_at":{"type":"string"},"first_requested_for_fulfillment_at":{"type":"string"},"fulfillment_exception_order_rejected_at":{"type":"string"},"requested_for_fulfillment_automatically":{"type":"boolean"},"fulfillment_exception_order_rejected_reason":{"type":"string"}}},supportedSyncModes=[full_refresh, incremental],sourceDefinedCursor=<null>,defaultCursorField=[],sourceDefinedPrimaryKey=[],namespace=analytics_zone,additionalProperties={}],syncMode=full_refresh,cursorField=[],destinationSyncMode=overwrite,primaryKey=[],additionalProperties={}]],additionalProperties={}],failures=[io.airbyte.config.FailureReason@755f2e9[failureOrigin=destination,failureType=system_error,internalMessage=redis.clients.jedis.exceptions.JedisMovedDataException: MOVED 480 10.4.12.55:6379,externalMessage=Something went wrong in the connector. See the logs for more details.,metadata=io.airbyte.config.Metadata@582da7d1[additionalProperties={attemptNumber=0, jobId=24419, from_trace_message=true}],stacktrace=redis.clients.jedis.exceptions.JedisMovedDataException: MOVED 480 10.4.12.55:6379
	at redis.clients.jedis.Protocol.processError(Protocol.java:121)
	at redis.clients.jedis.Protocol.process(Protocol.java:173)
	at redis.clients.jedis.Protocol.read(Protocol.java:227)
	at redis.clients.jedis.Connection.readProtocolWithCheckingBroken(Connection.java:352)
	at redis.clients.jedis.Connection.getIntegerReply(Connection.java:294)
	at redis.clients.jedis.Jedis.incr(Jedis.java:772)
	at io.airbyte.integrations.destination.redis.RedisHCache.insert(RedisHCache.java:30)
	at io.airbyte.integrations.destination.redis.RedisMessageConsumer.acceptTracked(RedisMessageConsumer.java:66)
	at io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer.accept(FailureTrackingAirbyteMessageConsumer.java:50)
	at io.airbyte.integrations.base.IntegrationRunner.consumeWriteStream(IntegrationRunner.java:194)
	at io.airbyte.integrations.base.IntegrationRunner.lambda$runConsumer$4(IntegrationRunner.java:203)
	at io.airbyte.integrations.base.IntegrationRunner.watchForOrphanThreads(IntegrationRunner.java:232)
	at io.airbyte.integrations.base.IntegrationRunner.runConsumer(IntegrationRunner.java:202)
	at io.airbyte.integrations.base.IntegrationRunner.lambda$runInternal$1(IntegrationRunner.java:165)
	at io.airbyte.integrations.base.sentry.AirbyteSentry.executeWithTracing(AirbyteSentry.java:54)
	at io.airbyte.integrations.base.sentry.AirbyteSentry.executeWithTracing(AirbyteSentry.java:38)
	at io.airbyte.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.java:165)
	at io.airbyte.integrations.base.IntegrationRunner.run(IntegrationRunner.java:107)
	at io.airbyte.integrations.destination.redis.RedisDestination.main(RedisDestination.java:24)
,retryable=<null>,timestamp=1656694689164], io.airbyte.config.FailureReason@3569da5f[failureOrigin=destination,failureType=<null>,internalMessage=io.airbyte.workers.general.DefaultReplicationWorker$DestinationException: Destination process end of stream notification failed,externalMessage=Something went wrong within the destination connector,metadata=io.airbyte.config.Metadata@2bf17420[additionalProperties={attemptNumber=0, jobId=24419}],stacktrace=java.util.concurrent.CompletionException: io.airbyte.workers.general.DefaultReplicationWorker$DestinationException: Destination process end of stream notification failed
	at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:315)
	at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:320)
	at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1807)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: io.airbyte.workers.general.DefaultReplicationWorker$DestinationException: Destination process end of stream notification failed
	at io.airbyte.workers.general.DefaultReplicationWorker.lambda$getReplicationRunnable$6(DefaultReplicationWorker.java:365)
	at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804)
	... 3 more
Caused by: java.io.IOException: Stream closed
	at java.base/java.lang.ProcessBuilder$NullOutputStream.write(ProcessBuilder.java:445)
	at java.base/java.io.OutputStream.write(OutputStream.java:162)
	at java.base/java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:81)
	at java.base/java.io.BufferedOutputStream.flush(BufferedOutputStream.java:142)
	at java.base/sun.nio.cs.StreamEncoder.implFlush(StreamEncoder.java:320)
	at java.base/sun.nio.cs.StreamEncoder.flush(StreamEncoder.java:160)
	at java.base/java.io.OutputStreamWriter.flush(OutputStreamWriter.java:248)
	at java.base/java.io.BufferedWriter.flush(BufferedWriter.java:257)
	at io.airbyte.workers.internal.DefaultAirbyteDestination.notifyEndOfInput(DefaultAirbyteDestination.java:98)
	at io.airbyte.workers.general.DefaultReplicationWorker.lambda$getReplicationRunnable$6(DefaultReplicationWorker.java:363)
	... 4 more
,retryable=<null>,timestamp=1656694689334], io.airbyte.config.FailureReason@31b090d[failureOrigin=destination,failureType=<null>,internalMessage=io.airbyte.workers.general.DefaultReplicationWorker$DestinationException: Destination process exited with non-zero exit code 1,externalMessage=Something went wrong within the destination connector,metadata=io.airbyte.config.Metadata@36bd3590[additionalProperties={attemptNumber=0, jobId=24419}],stacktrace=java.util.concurrent.CompletionException: io.airbyte.workers.general.DefaultReplicationWorker$DestinationException: Destination process exited with non-zero exit code 1
	at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:315)
	at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:320)
	at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1807)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: io.airbyte.workers.general.DefaultReplicationWorker$DestinationException: Destination process exited with non-zero exit code 1
	at io.airbyte.workers.general.DefaultReplicationWorker.lambda$getDestinationOutputRunnable$7(DefaultReplicationWorker.java:410)
	at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804)
	... 3 more
,retryable=<null>,timestamp=1656694689319]]]

sajarin · July 6, 2022, 3:47pm

Hi @israel.mendes welcome to the community and thanks for your post. Based on the logs, this seems to be an issue with the Redis destination. Could you describe how you configured your Redis instance/cluster? Also the logs indicate that there were 23 records synced. How many total records are you trying to sync to Redis from Redshift? Hope we can get this resolved soon!

israel.mendes · July 6, 2022, 9:12pm

Hey @sajarin

About your questions:

Could you describe how you configured your Redis instance/cluster?
- I’m using a simple Redis to test the connection. The configuration: Redis in AWS MemoryDB, cluster with db.t4g.small node type. Security group open to all IPs, cluster located in a private subnet, the same used by Airbyte. When I configured the destination, the authentication works fine between Airbyte EC2 and Redis.
How many total records are you trying to sync to Redis from Redshift?
- The table that I used in that connection only contains 23 records, which appear to be all in Redis. The problem seems that the connection fails for no clear reason.

marcosmarxm · August 1, 2022, 7:28pm

Sorry the long delay Israel, I created one issue in Github https://github.com/airbytehq/airbyte/issues/15169 looks the destination is not closing the ingestion step. Any update I’ll return to you.

Topic		Replies	Views
SFTP Failing to Redshift Connector Questions & Issues data-loading	4	320	July 14, 2022
Destination-redshift 0.3.32 issues with SUPER data type Connector Questions & Issues destination-redshift , data-loading	3	370	July 14, 2022
Redshift destination setup keeps timing out Connector Questions & Issues destination-redshift	2	877	March 16, 2023
Redshift destination failing with the latest version 0.3.32 rollback to 0.3.28 fixes issue Connector Questions & Issues data-loading , connectors	4	277	July 14, 2022
Redshift Source Discovering schema failed Connector Questions & Issues source-redshift , getting-started , data-loading	6	799	October 24, 2022

Destination Redis: "JedisMovedDataException" when running Redshift to Redis connection

Related topics