| 109 | } |
| 110 | |
| 111 | func (dp *dynamoProcess) doRequest(inp *dynamodb.BatchWriteItemInput) error { |
| 112 | var err error |
| 113 | backoff := awsutils.DefaultBackoff |
| 114 | deadline := time.Now().Add(dp.maxbackoff) |
| 115 | |
| 116 | for time.Now().Before(deadline) { |
| 117 | var resp *dynamodb.BatchWriteItemOutput |
| 118 | |
| 119 | req, resp := dp.db.BatchWriteItemRequest(inp) |
| 120 | err := req.Send() |
| 121 | if err != nil && !req.IsErrorRetryable() && !req.IsErrorThrottle() { |
| 122 | // If we got err here, it means that the whole batch failed with a |
| 123 | // non-retryable error. This must be something "permanent" like |
| 124 | // wrong table name. Just exit. |
| 125 | // FIXME: maybe we should differentiate batch mode vs daemon mode, |
| 126 | // and never exits in daemon mode? |
| 127 | log.Fatal(err) |
| 128 | return err |
| 129 | } |
| 130 | |
| 131 | // If all items were processed, exit |
| 132 | if len(resp.UnprocessedItems) == 0 { |
| 133 | return nil |
| 134 | } |
| 135 | |
| 136 | // There are unprocessed items. This is possibly a transient error (usually |
| 137 | // throughput error), so schedule a retry after a backoff. |
| 138 | time.Sleep(backoff.Duration()) |
| 139 | inp = &dynamodb.BatchWriteItemInput{ |
| 140 | RequestItems: resp.UnprocessedItems, |
| 141 | } |
| 142 | atomic.AddInt64(&dp.Stats.TotalRetries, 1) |
| 143 | } |
| 144 | |
| 145 | return err |
| 146 | } |
| 147 | |
| 148 | func (dp *dynamoProcess) run() { |
| 149 | for req := range dp.in { |